Розробка Workflow-двигуна на Apache Airflow під ключ

Розробка Workflow-двигуна на Apache Airflow під ключ

У вас зростає обсяг даних, але ручний запуск скриптів і cron'ів вже не справляється? Ми стикалися з ситуацією, коли N+1 запитів в ETL-пайплайні призводив до годинних затримок, а моніторинг був відсутній. Пайплайн завантаження 5 млн замовлень на LocalExecutor виконувався 4 години, а при пікових навантаженнях час простою сягав 30 хвилин. Ми вирішуємо цю проблему за допомогою Apache Airflow — зрілої платформи для оркестрації та автоматизації процесів обробки даних. Наші інженери побудували десятки DAG-пайплайнів для компаній з e-commerce, fintech та логістики. Кожен пайплайн проходить навантажувальне тестування: типовий DAG обробляє до 10 млн записів за ніч, вкладаючись в SLA при 99.9% часу виконання. Стабільність підтверджена 50+ проектами. Отримайте консультацію по вашому проекту — оцінимо за 1 день.

Чому варто обрати KubernetesExecutor?

Airflow оптимізований для batch-обробки даних:

• ETL/ELT пайплайни (PostgreSQL → трансформація → Data Warehouse)
• Щоденні звіти та вивантаження
• ML-пайплайни (підготовка даних → навчання → деплой моделі)
• Періодичні агрегації та синхронізації

Якщо у вас подійно-керовані бізнес-процеси з human tasks — варто придивитися до Temporal або Camunda. Airflow не вміє чекати користувача годинами. Але для data-інженерів це найкращий вибір: Airflow в 3-5 разів швидший за Temporal в batch-сценаріях, а також краще підходить для ETL-пайплайнів.

Переваги KubernetesExecutor

KubernetesExecutor дає ізоляцію на рівні завдань: кожне запускається в окремому Pod з власними CPU та пам'яттю. При пікових навантаженнях Kubernetes автоматично піднімає Pod'и, а після — утилізує. Ми використовуємо цей підхід у production і вважаємо його стандартом для сучасних data-пайплайнів.

Як Airflow пришвидшує обробку даних?

При піковому навантаженні KubernetesExecutor масштабує ресурси горизонтально: замість однієї ноди — кластер з 10+ Pod'ів. Час виконання ETL-пайплайну скорочується в 3-5 разів порівняно з LocalExecutor. Наприклад, пайплайн завантаження 5 млн замовлень на LocalExecutor виконується 4 години, на KubernetesExecutor — 50 хвилин. Різниця очевидна.

Побудова ETL-пайплайну: розгорнутий кейс

Розглянемо приклад для інтернет-магазину: щоденне завантаження замовлень з PostgreSQL в DWH (PostgreSQL) з трансформацією та агрегацією. Нижче — DAG, який ми розгорнули замовнику за 2 дні.

Встановлення через Helm

helm repo add apache-airflow https://airflow.apache.org
helm upgrade --install airflow apache-airflow/airflow \
  --namespace airflow \
  --create-namespace \
  --set executor=KubernetesExecutor \
  --set postgresql.enabled=true \
  --set redis.enabled=true \
  --values airflow-values.yaml

---
# airflow-values.yaml
aiflow:
  image:
    repository: apache/airflow
    tag: 2.8.0
  config:
    AIRFLOW__CORE__DAGS_FOLDER: /opt/airflow/dags
    AIRFLOW__CORE__MAX_ACTIVE_RUNS_PER_DAG: "3"
    AIRFLOW__SCHEDULER__MIN_FILE_PROCESS_INTERVAL: "30"
  dags:
    persistence:
      enabled: true
    gitSync:
      enabled: true
      repo: https://github.com/yourcompany/airflow-dags.git
      branch: main
      subPath: dags/
---

DAG — приклад ETL-пайплайну

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.postgres.operators.postgres import PostgresOperator
from airflow.providers.postgres.hooks.postgres import PostgresHook
from datetime import datetime, timedelta
import pandas as pd

default_args = {
    'owner': 'data-team',
    'depends_on_past': False,
    'start_date': datetime(2024, 1, 1),
    'retries': 2,
    'retry_delay': timedelta(minutes=5),
}

with DAG(
    'daily_orders_etl',
    default_args=default_args,
    schedule_interval='0 2 * * *',
    catchup=False,
    tags=['etl', 'orders'],
    description='Завантаження та трансформація замовлень в DWH',
) as dag:

    def extract_orders(**context):
        hook = PostgresHook(postgres_conn_id='production_db')
        ds = context['ds']
        df = hook.get_pandas_df(f"""
            SELECT o.id, o.customer_id, o.total, o.status, o.created_at,
                   c.email, c.country
            FROM orders o
            JOIN customers c ON c.id = o.customer_id
            WHERE o.created_at::date = '{ds}'
              AND o.status IN ('paid', 'shipped', 'delivered')
        """)
        context['ti'].xcom_push(key='orders_count', value=len(df))
        df.to_parquet(f'/tmp/orders_{ds}.parquet')
        return len(df)

    def transform_orders(**context):
        ds = context['ds']
        df = pd.read_parquet(f'/tmp/orders_{ds}.parquet')
        df['order_date'] = pd.to_datetime(df['created_at']).dt.date
        df['revenue_usd'] = df['total'] / 100
        df['is_international'] = df['country'] != 'RU'
        df['customer_tier'] = df['revenue_usd'].apply(
            lambda x: 'vip' if x >= 500 else 'regular'
        )
        df.to_parquet(f'/tmp/orders_transformed_{ds}.parquet')

    def load_to_dwh(**context):
        ds = context['ds']
        df = pd.read_parquet(f'/tmp/orders_transformed_{ds}.parquet')
        hook = PostgresHook(postgres_conn_id='datawarehouse')
        engine = hook.get_sqlalchemy_engine()
        df.to_sql('fact_orders', engine, schema='dwh', if_exists='append', index=False, method='multi', chunksize=1000)

    aggregate_metrics = PostgresOperator(
        task_id='aggregate_metrics',
        postgres_conn_id='datawarehouse',
        sql="""
            INSERT INTO dwh.daily_metrics (date, total_revenue, orders_count, avg_order)
            SELECT '{{ ds }}'::date, SUM(revenue_usd), COUNT(*), AVG(revenue_usd)
            FROM dwh.fact_orders
            WHERE order_date = '{{ ds }}'
            ON CONFLICT (date) DO UPDATE SET
                total_revenue = EXCLUDED.total_revenue,
                orders_count = EXCLUDED.orders_count,
                avg_order = EXCLUDED.avg_order;
        """,
    )

    extract = PythonOperator(
        task_id='extract_orders',
        python_callable=extract_orders
    )
    transform = PythonOperator(
        task_id='transform_orders',
        python_callable=transform_orders
    )
    load = PythonOperator(
        task_id='load_to_dwh',
        python_callable=load_to_dwh
    )

    extract >> transform >> load >> aggregate_metrics

Паралельне виконання та Sensors

from airflow.utils.task_group import TaskGroup

with TaskGroup('process_regions') as process_regions:
    for region in ['EU', 'US', 'APAC']:
        PythonOperator(
            task_id=f'process_{region.lower()}',
            python_callable=process_region_data,
            op_kwargs={'region': region}
        )

extract >> process_regions >> aggregate_all

Для очікування зовнішніх подій використовуємо Sensors: FileSensor для файлів, HttpSensor для API. Це стандартний патерн для production.

KubernetesExecutor в дії на Apache Airflow

---
executor_config = {
    'KubernetesExecutor': {
        'request_memory': '2Gi',
        'request_cpu': '500m',
        'limit_memory': '4Gi',
        'image': 'custom-airflow:2.8.0-pandas',
    }
}
heavy_transform = PythonOperator(
    task_id='heavy_transform',
    python_callable=transform_large_dataset,
    executor_config=executor_config
)

Обробка збоїв DAG

При збої завдання Airflow автоматично виконує повторні спроби (retries) з експоненційною затримкою. Ми налаштовуємо сповіщення в Telegram/Slack на кожен збій та тривале виконання. Додатково інтегруємо метрики в Prometheus/Grafana: дашборди показують час виконання, кількість успішних/невдалих завдань, завантаження ресурсів. Це дозволяє швидко реагувати на інциденти.

Процес роботи

1. Аналіз: вивчаємо ваші джерела даних, обсяги, SLA, поточні проблеми, налаштування Airflow та інфраструктуру.
2. Проектування: описуємо DAG'и, обираємо Executor, продумуємо обробку помилок та сповіщення.
3. Реалізація: пишемо код DAG'ів, конфіги Helm, CI/CD, тести.
4. Деплой: розгортаємо інфраструктуру (Kubernetes або Docker), налаштовуємо GitSync та моніторинг.
5. Тестування: прогоняємо backfill на історичних даних, перевіряємо коректність.
6. Документація та навчання: передаємо runbook, навчаємо вашу команду.
7. Підтримка: 2 тижні після запуску для стабілізації.

Що входить в роботу

• Розроблені DAG'и (Python-код, готовий до production).
• Конфігурація Airflow (values.yaml, змінні, connections).
• CI/CD pipeline для автоматичного деплою DAG'ів через Git.
• Docker-образ з залежностями (бібліотеки, драйвери).
• Документація: опис DAG'ів, інструкція щодо запуску, troubleshooting.
• Навчання команди (2-3 сесії).
• Моніторинг: сповіщення в Telegram/Slack, дашборди Grafana.

Терміни реалізації

Наш досвід та гарантії

У нас 10+ років досвіду в розробці data-інфраструктури. Ми реалізували 50+ проектів на Airflow для e-commerce, fintech та логістики. Середня uptime пайплайнів — 99.9%. Використовуємо best practices: GitSync, KubernetesExecutor, сповіщення, версіонування DAG'ів. Гарантуємо стабільну роботу пайплайнів та документацію рівня enterprise.

Ми готові розробити аналогічний пайплайн для ваших даних. Зв'яжіться з нами для оцінки — відповімо за 1 день. Або замовте консультацію — обговоримо деталі без зобов'язань.

Apache Airflow Documentation