Как организовать автосборку данных и обучение с помощью Airflow

Что проверяют

Понимаете ли вы Airflow как оркестратор, а не как место, где крутятся тяжёлые вычисления. Следом спросят, как вы передаёте данные между задачами и что будет при повторном запуске за прошлую дату.

Как отвечать

Опишите пайплайн как DAG из идемпотентных шагов, каждый из которых обрабатывает свой интервал данных:

• извлечение: задача выгружает данные между data_interval_start и data_interval_end и пишет партицию по дате в объектное хранилище или DWH. Повторный запуск перезаписывает ту же партицию, а не дописывает дубли;
• проверка качества: отдельная задача с Great Expectations, Pandera или простыми SQL-проверками. Упала проверка, обучение не стартует;
• признаки и обучение: тяжёлое выносится из воркера в KubernetesPodOperator, Spark или отдельный сервис обучения с GPU. Airflow только запускает и ждёт;
• оценка и регистрация: метрики сравниваются с текущей моделью, лучшая попадает в MLflow Model Registry. Ветвление через @task.branch или ShortCircuitOperator;
• передача данных: через XCom идут только пути и идентификаторы, сами датафреймы и модели лежат в хранилище.

Покажите современный синтаксис: TaskFlow API с @dag и @task, параметр schedule, потому что schedule_interval в Airflow 3 удалён. Там же стандартные операторы переехали в airflow.providers.standard, а Datasets стали Assets: DAG обучения можно запускать не по расписанию, а по обновлению ассета с данными. Для однотипных задач по списку моделей подходит dynamic task mapping через .expand().

Назовите эксплуатацию: retries и retry_delay, execution_timeout, on_failure_callback с уведомлением, пулы для ограничения параллельных тяжёлых задач, catchup=False для DAG, которым не нужна история.

Где спотыкаются

Гоняют обучение прямо в PythonOperator на воркере. Тяжёлый шаг занимает слот и память, и соседние DAG встают в очередь.

Кладут в XCom датафреймы и модели. База метаданных разрастается, а сериализация больших объектов рано или поздно ломается.

Пишут неидемпотентные задачи с datetime.now() вместо интервала запуска. Перезапуск или backfill за прошлую дату выгружает не те данные и дублирует строки.