Airflowdata engineeringETLorchestrationmodern data stack

Airflow для аналитика 2026: DAG patterns + best practices

2026-05-30 11 мин

Apache Airflow — стандарт оркестрации пайплайнов в 2026. Если ты dbt-моделишь данные, Airflow запускает dbt по расписанию + ходит за свежими файлами + рассылает алерты в Slack + сохраняет результаты тестов. Этот гайд — практические DAG-паттерны без воды.

Покажу как писать идемпотентные DAG, использовать TaskFlow API (2.0+), грамотно работать с sensors и динамическими DAG-ами. Всё с примерами из аналитики маркетплейса.

Что узнаешь

Шаг 1: Airflow за 5 минут — концепции

DAG (Directed Acyclic Graph) — расписание задач + их зависимости.

Task — единица работы (запустить SQL, скопировать файл).

Scheduler — следит за расписаниями DAGов и кладёт их в очередь.

Worker — выполняет tasks из очереди.

Executor — стратегия выполнения (Local, Celery, Kubernetes).

Source DBs → Airflow Scheduler (cron) → Workers
                   ↓
            DAGs запускают:
            dbt run + tests
            S3 → Snowflake load
            Notebook → отчёт в Slack

Подборка 10 Airflow-вопросов с собесов — паттерны idempotency, scheduling и sensors.

Шаг 2: TaskFlow API (Airflow 2.0+) — современный синтаксис

Старый PythonOperator:

def extract(ds, **kwargs):
    return {'rows': 1000}

def transform(ti, **kwargs):
    data = ti.xcom_pull(task_ids='extract')
    return {'processed': data['rows'] * 2}

with DAG('old_style', schedule='@daily') as dag:
    e = PythonOperator(task_id='extract', python_callable=extract)
    t = PythonOperator(task_id='transform', python_callable=transform)
    e >> t

TaskFlow API — короче в 3 раза:

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

@dag(schedule='@daily', start_date=datetime(2024, 1, 1), catchup=False, tags=['etl'])
def etl_pipeline():
    @task
    def extract():
        return {'rows': 1000}

    @task
    def transform(data: dict):
        return {'processed': data['rows'] * 2}

    @task
    def load(data: dict):
        print(f"Loaded {data['processed']} rows")

    load(transform(extract()))

etl_pipeline()

XCom передаётся автоматически через return value → assignment.

Шаг 3: Idempotency — главный паттерн

Idempotent task: запустить N раз = тот же результат. Безопасный retry без дублирования.

Bad (не идемпотентный):

@task
def load_revenue(ds):
    df = pd.read_csv(f's3://raw/events_{ds}.csv')
    df.to_sql('daily_revenue', engine, if_exists='append')   # ❌ retry дублирует

Good (идемпотентный):

@task
def load_revenue(ds):
    df = pd.read_csv(f's3://raw/events_{ds}.csv')
    with engine.begin() as conn:
        conn.execute(f"DELETE FROM daily_revenue WHERE day = '{ds}'")
        df.to_sql('daily_revenue', conn, if_exists='append', index=False)

Лучший вариант — UPSERT/MERGE:

@task
def load_revenue(ds):
    sql = f"""
    INSERT INTO daily_revenue (day, revenue)
    SELECT '{ds}', SUM(amount)
    FROM events
    WHERE day = '{ds}'
    GROUP BY day
    ON CONFLICT (day) DO UPDATE SET revenue = EXCLUDED.revenue;
    """
    engine.execute(sql)

Использование {{ ds }} (execution_date logical date в YYYY-MM-DD) — Airflow подставит правильную дату при backfill и retry.

Шаг 4: Sensors — ожидание условий

Sensor ждёт пока условие станет True (файл появился, partition готов, API доступен).

Modes:

from airflow.sensors.filesystem import FileSensor

wait_for_file = FileSensor(
    task_id='wait_for_csv',
    filepath='/data/incoming/sales_{{ ds }}.csv',
    poke_interval=60,           # проверять каждую минуту
    timeout=3600,               # ждать максимум 1 час
    mode='reschedule',          # освобождать worker между проверками
)

ExternalTaskSensor — ждать task из другого DAG:

from airflow.sensors.external_task import ExternalTaskSensor

wait_upstream = ExternalTaskSensor(
    task_id='wait_upstream_dag',
    external_dag_id='upstream_etl',
    external_task_id='load_final',
    timeout=7200,
    mode='reschedule',
)

Smart Sensors deprecated в 2.4+ → используй deferrable operators (новая модель async).

Шаг 5: Dynamic Task Mapping (Airflow 2.3+)

Один task description → N исполнений на runtime данных:

@dag(schedule='@daily', start_date=datetime(2024, 1, 1))
def per_country_etl():
    @task
    def list_countries():
        return ['ru', 'us', 'br', 'de', 'in']  # из DB или config

    @task
    def process_country(country: str):
        # Обработка для конкретной страны
        print(f"Processing {country}")
        return f"done_{country}"

    @task
    def summarize(results: list):
        print(f"Total countries: {len(results)}")

    countries = list_countries()
    results = process_country.expand(country=countries)
    summarize(results)

При runtime Airflow создаст 5 task-инстансов process_country (по одному на страну), параллельно или последовательно в зависимости от max_active_tis_per_dag.

Шаг 6: Error handling + alerts

on_failure_callback — слать в Slack/PagerDuty при падении:

def slack_alert(context):
    task = context['task'].task_id
    dag = context['dag'].dag_id
    msg = f"⚠️ {dag}.{task} FAILED. Log: {context['task_instance'].log_url}"
    send_to_slack(msg)

@dag(
    schedule='@daily',
    on_failure_callback=slack_alert,    # на падение всего DAG
    default_args={
        'on_failure_callback': slack_alert,  # на падение task
        'retries': 3,
        'retry_delay': timedelta(minutes=5),
        'sla': timedelta(hours=2),
        'email_on_failure': False,
    }
)
def critical_etl():
    ...

SLA misses автоматом регистрируются если task не завершился в указанный timeframe — sla_miss_callback шлёт уведомление.

Шаг 7: Best practices в проде

from airflow.utils.task_group import TaskGroup

with TaskGroup('etl_block') as etl_group:
    extract() >> transform() >> load()

heavy_task = PythonOperator(
    task_id='heavy',
    python_callable=my_func,
    pool='snowflake_pool',
    pool_slots=2,
)

FAQ

Airflow vs Prefect vs Dagster?

Airflow — мейн-стрим (большая комьюнити, dbt integration, K8s executor). Prefect — современный API, лучше для ML pipelines. Dagster — software-defined assets, асset-centric model. В РФ в 99% случаев Airflow.

Сколько DAG норм для одного instance?

До 500-1000 DAG на standalone Scheduler без проблем. После — нужны несколько Scheduler-инстансов или HA setup. На очень больших instance (3000+) лучше разделить на logical clusters.

Schedule '@daily' vs '0 0 * * *'?

Эквивалентны: оба = midnight UTC каждый день. Cron-выражения дают точный контроль. @daily короче и понятнее.

catchup=True — кому нужен?

Для bulk-backfill при первом запуске нового DAG. После — переключай на False, иначе любое изменение start_date запустит сотни runs.

Dependence на dbt?

Используй BashOperator (dbt run --select model_name) или Astronomer Cosmos (https://github.com/astronomer/astronomer-cosmos) — генерит Airflow tasks из dbt manifest автоматически.

Что дальше

Источники

Airflow — это контроль над тысячами пайплайнов. Открой SQL-тренажёр и качай SQL под dbt-модели которые Airflow будет крутить.

Прокачай SQL для пайплайнов
480+ SQL-задач, 600+ заданий из реальных собесов, AI-разбор. Бесплатный старт.
Открыть тренажёр →

Похожие статьи