Airflow Best Practices: Динамические DAGи и TaskFlow API

Содержание

Динамическая генерация DAG-ов в Apache Airflow
Шаг 1. Конфигурация (dags/config/tables.yaml)
Шаг 2. Генератор (dags/dynamic_dag_generator.py)
TaskFlow API - новый стиль написания кода в AirFlow
Мониторинг и Алертинг - что делать когда все сломалось
Troubleshooting: Ошибки, которые убивают прод
Как Cursor помогает в рефакторинге
Финал цикла
Использованные референсы и материалы
Полный перечень статей Бесплатного курса "Apache Airflow для начинающих"

Представьте, что вы работаете в e-commerce. У вас есть 50 таблиц в Postgres (заказы, товары, пользователи, отзывы…), и каждую из них нужно переливать в ClickHouse по одной и той же схеме: Скачать -> Очистить -> Загрузить.

Новичок создаст 50 файлов: dag_orders.py, dag_users.py, dag_items.py… В каждом файле будет одинаковый код, отличающийся только названием таблицы. Это антипаттерн. Это нарушает принцип DRY (Don’t Repeat Yourself).

Apache Airflow — это Python. А значит, мы можем использовать циклы, конфиги и метапрограммирование, чтобы Airflow сам создавал себе задачи.

Динамическая генерация DAG-ов в Apache Airflow

Концепция проста: мы создаем один файл конфигурации (JSON или YAML), где перечисляем наши таблицы, и один Python-скрипт, который в цикле читает этот конфиг и «печет» DAG-и как пирожки.

Шаг 1. Конфигурация (dags/config/tables.yaml)

orders:
  schedule: "@daily"
  source_table: "public.orders"
  dest_table: "analytics.orders"

users:
  schedule: "@weekly"
  source_table: "public.users"
  dest_table: "analytics.users"

Шаг 2. Генератор (dags/dynamic_dag_generator.py)

Тут есть нюанс. Airflow ищет объекты DAG в глобальном пространстве имен (globals()).

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
import yaml
import os

# Функция-фабрика, которая создает DAG
def create_dag(dag_id, schedule, source, dest):
    
    # Обычная функция обработки
    def process_data():
        print(f"Переливаем данные из {source} в {dest}")
        # Тут может быть вызов вашего S3Hook или ClickHouseHook
   generated_dag = DAG(
        dag_id=dag_id,
        schedule_interval=schedule,
        start_date=datetime(2023, 1, 1),
        catchup=False
    )

    with generated_dag:
        t1 = PythonOperator(
            task_id="process_table",
            python_callable=process_data
        )

    return generated_dag

# Читаем конфиг
config_path = os.path.join(os.path.dirname(__file__), "config", "tables.yaml")
with open(config_path) as f:
    config = yaml.safe_load(f)
# Магия генерации
for table_name, params in config.items():
    dag_id = f"etl_{table_name}"
    
    # Создаем объект DAG
    dag_obj = create_dag(
        dag_id=dag_id,
        schedule=params['schedule'],
        source=params['source_table'],
        dest=params['dest_table']
    )
    
    # ВАЖНО: Кладем DAG в globals(), чтобы Airflow его увидел
    globals()[dag_id] = dag_obj

Теперь, если вы добавите новую таблицу в YAML-файл, Airflow автоматически создаст новый DAG при следующем сканировании папки. Никакой копипасты.

Apache Airflow для инженеров данных

Код курса

AIRF

Ближайшая дата курса

23 марта, 2026

Продолжительность

24 ак.часов

Стоимость обучения

76 800

TaskFlow API — новый стиль написания кода в AirFlow

Если вы посмотрите на наши предыдущие статьи, мы везде использовали классический стиль: PythonOperator, op_kwargs, ручной xcom_pull. Это рабочий, но многословный подход (boilerplate code).

Начиная с версии 2.0, Airflow представил TaskFlow API. Это использование декораторов @task и @dag.

Было (Старый стиль):

def extract(**kwargs):
    return {"data": [1, 2, 3]}

def transform(**kwargs):
    ti = kwargs['ti']
    data = ti.xcom_pull(task_ids='extract')
    return [x * 2 for x in data['data']]

t1 = PythonOperator(task_id='extract', python_callable=extract)
t2 = PythonOperator(task_id='transform', python_callable=transform)
t1 >> t2

Стало (TaskFlow API):

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

@dag(start_date=datetime(2023, 1, 1), schedule=None, catchup=False)
def taskflow_etl():

    @task
    def extract():
        return [1, 2, 3]

    @task
    def transform(data_list):
        # Airflow сам поймет, что data_list пришел из предыдущей задачи
        # и сам сделает xcom_pull под капотом
        return [x * 2 for x in data_list]

    @task
    def load(data_list):
        print(f"Loading: {data_list}")
  # Задаем порядок выполнения как вызов функций
    raw_data = extract()
    clean_data = transform(raw_data)
    load(clean_data)

# Инициализируем DAG
my_dag = taskflow_etl()

Это делает код чище, понятнее и больше похожим на обычный Python-скрипт. Airflow берет на себя всю «грязную работу» по передаче данных.

Мониторинг и Алертинг — что делать когда все сломалось

Никто не смотрит в монитор Airflow 24/7. Если DAG упал ночью, вы должны узнать об этом сразу. В Airflow есть механизм Callbacks (обратных вызовов). Мы можем повесить функцию на событие on_failure_callback.

Пример: Отправка уведомления в Telegram

Сначала напишем функцию отправки (понадобится pip install requests):

import requests

def send_telegram_alert(context):
    token = "ВАШ_БОТ_ТОКЕН"
    chat_id = "ВАШ_CHAT_ID"
    
    # context содержит информацию об упавшей задаче
    task_instance = context.get('task_instance')
    dag_id = task_instance.dag_id
    task_id = task_instance.task_id
    log_url = task_instance.log_url
    exception = context.get('exception')

    msg = f"""
    🔴 **ALARM! Task Failed**
    **DAG:** {dag_id}
    **Task:** {task_id}
    **Error:** {str(exception)[:100]}...
    [See Logs]({log_url})
    """
    
    url = f"https://api.telegram.org/bot{token}/sendMessage"
    requests.post(url, data={'chat_id': chat_id, 'text': msg, 'parse_mode': 'Markdown'})


Теперь подключим её ко всем задачам в DAG через default_args:
default_args = {
    'owner': 'airflow',
    'retries': 1,
    'on_failure_callback': send_telegram_alert # <--- Магия здесь
}

with DAG(..., default_args=default_args) as dag:
    # Теперь, если любая задача упадет, Airflow вызовет функцию алерта
    ...

Apache Airflow для инженеров данных

Код курса

AIRF

Ближайшая дата курса

23 марта, 2026

Продолжительность

24 ак.часов

Стоимость обучения

76 800

Troubleshooting: Ошибки, которые убивают прод

Ошибка 1: Тяжелый код на верхнем уровне (Top-Level Code) Планировщик Airflow сканирует папку dags каждые 30 секунд (по умолчанию). Он открывает каждый файл и исполняет код, чтобы найти объекты DAG.

Плохо: Делать запросы к БД или API вне функций операторов.

<!-- end list -->
# НИКОГДА ТАК НЕ ДЕЛАЙТЕ
my_data = requests.get("https://api.example.com").json() # <-- Этот код будет выполняться каждые 30 сек планировщиком!

with DAG(...) as dag:
    ...

Последствия: Планировщик зависнет, CPU загрузится на 100%, новые задачи перестанут запускаться.
Решение: Весь тяжелый код только внутри def my_func(): или внутри операторов.

Ошибка 2: Передача Гигабайтов через XCom TaskFlow API делает передачу данных легкой (extract() >> transform()), и возникает соблазн передавать огромные датафреймы.

Проблема: XCom пишет данные в базу метаданных (Postgres). Она не предназначена для хранения блобов по 500 МБ.
Последствия: База Airflow распухает, интерфейс тормозит, задачи падают с ошибкой базы.
Решение: Передавайте через XCom только пути к файлам в S3 (s3://bucket/data.csv), а не сами данные.

Ошибка 3: Зомби-задачи Иногда задача помечена как Running, но воркер уже умер или был перезагружен.

Решение: Настройте параметры scheduler_zombie_task_threshold. Airflow периодически проверяет, жив ли процесс, и если нет — помечает задачу как Failed, чтобы сработал Retry.

Как Cursor помогает в рефакторинге

Миграция старого кода на новые рельсы — идеальная задача для AI.

Сценарий: У вас есть старый DAG на PythonOperator, и вы хотите переписать его на TaskFlow API. Промпт для Cursor: «Перепиши этот код DAG-а, используя современный стиль Airflow TaskFlow API (декораторы @task и @dag). Убедись, что передача данных между задачами происходит через возвращаемые значения, а не через явный xcom_pull.»

Сценарий: Генерация конфига для динамических дагов. Промпт: «У меня есть список из 10 таблиц: [table1, table2…]. Создай YAML-конфиг файл для генерации DAG-ов Airflow, где для каждой таблицы будет указано расписание и имя целевой таблицы. Также напиши Python-код генератора, который читает этот YAML.»

Финал цикла

Поздравляю! Мы прошли путь от «Hello World» в консоли до построения сложной, отказоустойчивой платформы данных.

Чему мы научились за 10 статей:

Что дальше? Дальше — только практика. Apache Airflow — это стандарт индустрии. Знание того, как он работает «под капотом» (о чем мы говорили в каждой статье), делает вас не просто пользователем инструмента, а инженером, способным проектировать надежные системы.

Спасибо, что были с нами в этом путешествии. Удачных пайплайнов и зеленых дагов! 🟢

Все финальные варианты кода Dags и конфигурационных файлов лежат в репозитории на GitHub как и обычно.

Использованные референсы и материалы

Airflow Best Practices Guide
https://airflow.apache.org/docs/apache-airflow/stable/best-practices.html
«Библия» инженера: что делать, а чего категорически избегать в продакшене.
Tutorial on the TaskFlow API
https://airflow.apache.org/docs/apache-airflow/stable/tutorial/taskflow.html
Как писать современные и чистые DAG-и через декораторы @task.
Logging and Monitoring Architecture
https://airflow.apache.org/docs/apache-airflow/stable/administration-and-deployment/logging-monitoring/callbacks.html
Как настроить алерты и коллбэки (on_failure_callback) для отслеживания ошибок.