Проектування архітектури Data Pipeline для AI

Проектування архітектури Data Pipeline для AI

Data Pipeline для AI — це система, яка трансформує сирові дані з різних джерел у готові до навчання датасети та ознаки для інференсу. Правильна архітектура визначає масштабованість, надійність та швидкість ітерації команди ML.

Типи AI Data Pipelines

Batch Pipeline — обробка даних великими порціями за розкладом. Підходить для навчання моделей, підготовки навчальних вибірок, щоденних рекомендацій.

Streaming Pipeline — обробка подій у реальному часі. Підходить для real-time рекомендацій, виявлення шахрайства, динамічного ціноутворення.

Lambda Architecture — комбінація batch та streaming. Batch шар забезпечує точність, speed layer — актуальність.

Kappa Architecture — все через streaming, batch — це просто переовідтворення історичних подій. Простіше оперативно, потребує потужнішого streaming движка.

Компоненти архітектури

Data Sources:
├── Operational DBs (PostgreSQL, MySQL) → CDC (Debezium)
├── Event Streams (Kafka) → Direct consumption
├── File Storage (S3) → Batch ingestion
├── APIs (REST, GraphQL) → Connectors (Airbyte, Fivetran)
└── ML Feedback (predictions) → Kafka events

Processing Layer:
├── Batch: Apache Spark / dbt / pandas
├── Streaming: Apache Flink / Spark Structured Streaming
└── Feature computation: Feast / Tecton

Storage Layer:
├── Raw Zone (S3/GCS): вихідні дані без змін
├── Curated Zone (Delta Lake/Iceberg): очищені дані
├── Feature Store: готові ознаки для ML
└── ML Artifacts (S3 + DVC): датасети для навчання

Orchestration:
└── Apache Airflow / Prefect / Dagster

Інкрементальна обробка

from airflow.decorators import dag, task
from datetime import datetime, timedelta

@dag(
    schedule_interval='@hourly',
    start_date=datetime(2024, 1, 1),
    catchup=False,
    default_args={'retries': 2, 'retry_delay': timedelta(minutes=5)}
)
def user_features_pipeline():

    @task
    def extract_events(execution_date=None):
        # Інкрементальне завантаження: лише нові події
        watermark = get_watermark('user_events')
        events = clickhouse.query(
            "SELECT * FROM user_events WHERE event_time > %(watermark)s",
            {'watermark': watermark}
        )
        update_watermark('user_events', events['event_time'].max())
        return events.to_parquet()

    @task
    def compute_features(events_path: str):
        events = pd.read_parquet(events_path)

        # Обчислення ознак
        features = events.groupby('user_id').agg({
            'event_time': 'max',
            'event_type': 'count',
            'session_duration': ['mean', 'sum'],
        }).reset_index()

        features.columns = [
            'user_id', 'last_activity', 'event_count',
            'avg_session_duration', 'total_session_time'
        ]
        return features.to_parquet()

    @task
    def materialize_to_feature_store(features_path: str):
        features = pd.read_parquet(features_path)
        feast_store.write_to_online_store('user_features', features)
        feast_store.write_to_offline_store('user_features', features)

    events = extract_events()
    features = compute_features(events)
    materialize_to_feature_store(features)

pipeline = user_features_pipeline()

Якість даних у Pipeline

from great_expectations.core import ExpectationSuite

class DataQualityValidator:
    def __init__(self, suite_name: str):
        self.context = great_expectations.get_context()
        self.suite = self.context.get_expectation_suite(suite_name)

    def validate(self, df: pd.DataFrame) -> ValidationResult:
        validator = self.context.get_validator(
            batch_request=RuntimeBatchRequest(
                datasource_name="pandas_datasource",
                data_connector_name="runtime",
                data_asset_name="ml_features",
                runtime_parameters={"batch_data": df},
                batch_identifiers={"run_id": str(uuid.uuid4())}
            ),
            expectation_suite=self.suite
        )

        results = validator.validate()
        if not results.success:
            failed = [r for r in results.results if not r.success]
            raise DataQualityError(f"Validation failed: {failed}")

        return results

Обробка Schema Evolution

# Delta Lake підтримує schema evolution:
from delta import DeltaTable

DeltaTable.forPath(spark, "s3://bucket/user_features") \
    .toDF() \
    .mergeSchema(new_schema) \
    .write \
    .option("mergeSchema", "true") \
    .format("delta") \
    .mode("append") \
    .save("s3://bucket/user_features")

Моніторинг Pipeline

Ключові метрики: freshness (затримка даних відносно джерела), completeness (% очікуваних записів отримано), latency (час виконання кожного етапу), error rate. Сповіщення: дані не оновлювалися >N годин, uptime упав нижче порогу, кількість записів аномально відрізняється від очікуваного.

Типовий результат проектування: ML-команда отримує свіжі ознаки кожні 15-60 хвилин замість щоденних batch розрахунків, час підготовки нової навчальної вибірки зменшується з кількох годин до 10-15 хвилин.