Зависимости задач: оператор сдвига

Ночной пайплайн «Гонца» теперь выгружает заказы, курьеров и погоду. В прошлом уроке мы настроили расписание, а задачи всё ещё стоят в одну цепочку: погода ждёт курьеров, курьеры ждут заказы, хотя друг от друга они не зависят. Ещё две просьбы от команды: при сбое писать Отто в мессенджер, а временные файлы удалять при любом исходе, иначе за месяц диск сервера забивается мусором.

Кухня ресторана

На кухне повар сначала ставит вариться бульон, режет овощи и жарит мясо. Эти три дела не ждут друг друга, их делают три человека одновременно. Суп собирают, только когда готовы все три. А мыть посуду будут при любом исходе: даже если суп пригорел и его вылили, кастрюли сами не отмоются.

Если мясо подгорело, шеф идёт извиняться перед гостем. Это действие запускается не после успеха, а после провала.

Порядок «что после чего» и правило «при каком исходе» в Airflow задаются отдельно.

Порядок: >> и <<

Порядок задач задают операторами сдвига. a >> b читается «a, потом b». b << a значит то же самое, но так пишут редко.

extract_orders >> build_fact >> check

Список с любой стороны от >> даёт веер и сборку:

start >> [extract_orders, extract_couriers, extract_weather]   # веер
[extract_orders, extract_couriers, extract_weather] >> build_fact   # сборка

Длинную цепочку удобно писать функцией chain из airflow.sdk: chain(a, b, c, d) это то же, что a >> b >> c >> d.

С TaskFlow порядок часто получается сам: если задача получает результат другой задачи аргументом, Airflow сам ставит её после. Операторы сдвига нужны, когда данные не передаются, а порядок важен.

Правила запуска: trigger_rule

По умолчанию задача запускается, когда все её предшественники закончились успешно. Если хоть один упал, задача не запускается и получает состояние upstream_failed. Это правило называется all_success.

Другое правило задаётся параметром trigger_rule. Самые нужные:

Правило	Когда задача запускается	Пример
all_success	все предшественники успешны (по умолчанию)	собрать факт после трёх выгрузок
all_done	все закончились, неважно как	удалить временные файлы
one_failed	хотя бы один упал	написать Отто о сбое
none_failed	никто не упал, пропущенные не мешают	продолжить после необязательного шага

Если правило не выполнилось, задачу пропускают: она получает skipped. Например, оповещение с one_failed в спокойную ночь будет skipped.

@task(trigger_rule="all_done")
def cleanup_tmp(): ...

@task(trigger_rule="one_failed")
def notify_otto(): ...