Планирование даг выполняется в Airflow

Question

Получен общий запрос на Airflow

Можно ли запланировать файл dag на основе расписания другого файла dag.

Например, если у меня есть 2 dag, а именно dag1 и dag2.Я пытаюсь увидеть, могу ли я запускать dag2 каждый раз, когда dag1 успешен, иначе dag2 не запускается.Возможно ли это в Airflow?

Answer 1

Интервал расписания DAG должен быть определен как один из:

• cron расписание
• предустановка, например, '@once', '@hourly',и т. д.
• None*

* В случае использования нулевого расписания группа обеспечения доступности баз данных не будет запускаться автоматически и должна запускаться каким-либо образом.

Один из способовтриггер DAG должен использовать SubDAGs через SubDagOperator.Я думаю, что SubDAG - это, вероятно, лучший вариант для вашего варианта использования, учитывая, что вы хотите, чтобы второй DAG был запущен в результате успешного завершения первого DAG.SubDAGs имеют некоторые нюансы, как описано в документации.

SubDAG будет автоматически запускаться, если задание выполнится до ее успешного выполнения, и пропустить, если задание выполнено до ее сбоя / пропуска при условии, что вы используете ALL_SUCCESS или ONE_SUCCESSкак ваше правило триггера .

[Этот подход несколько похож на оператор TriggerDagRunOperator , который является еще одним вариантом, подробно описанным в ответе @ andscoop.]

Другой способ вызвать группу доступности баз данных - использовать внешний триггер .Эта идея более подробно обсуждается в этом ответе .

Answer 2

Вы хотите добавить TriggerDagRunOperator конец dag1 и установить расписание dag2 на None.

Кроме того, если вы хотите обрабатывать несколькослучаи для вывода dag1, вы можете добавить в BranchPythonOperator , чтобы создать несколько путей на основе его вывода.Например, вы можете настроить его либо на выполнение TriggerDagRunOperator в случае успеха, либо на слабое «Предупреждение! Ошибка задачи в DAG1!»с SlackAPIPostOperator , если выдается ошибка (или любая другая логика, которую вы хотите встроить).

Если вам не нужны множественные результаты, вы также можете просто использовать ShortCircuitOperator перед TriggerDagRunOperator для предотвращения его запуска на основе вывода dag1.

Answer 3

Да.Вы должны включить задачу в конце dag1, используя TriggerDagRunOperator .Убедитесь, что у этой задачи есть правило триггера , которое позволит ей запускаться только в том случае, если все остальные задачи в восходящем направлении успешно выполняются.

В другом ответе рекомендуются подзадачи, которые могут иметь некоторые странности, которые делают их не идеальными.в сложных условиях воздушного потока.