Как настроить DAG Airflow на ожидание завершения заданий Dataflow?

Question

У меня есть группа доступности базы данных, которая выполняет 3 конвейера потока данных.Я установил зависимость следующим образом:

a > b > c

Я установил следующие параметры по умолчанию:

default_dag_args = {
   'start_date': yesterday,
   'depends_on_past': True,
   'wait_for_downstream': True
}

Однако кажется, что все 3 конвейера планируются одновременно.Как настроить конвейер b на запуск только после завершения конвейера a?И точно так же конвейер c будет работать только после того, как конвейер b завершится?

Обновление:

Я изменил его на:

a >> b >> c

Теперь кажется, что a сработает и завершится, но б никогда не начинается.Группа DAG активна («Вкл»).Задача a все еще находится в состоянии «выполняется» в Airflow, но в потоке данных задание завершено.Как заставить Airflow распознать, что задание Dataflow завершено, и перейти к задаче b?

Answer 1

Используйте следующий флаг default_dag_args:

default_dag_args = {
   'start_date': yesterday,
   'depends_on_past': False,
   'wait_for_downstream': True
}

depends_on_past, чтобы фактически выполнить предыдущий запуск той же задачи.Поэтому, если предыдущий экземпляр задачи не удался, эта задача не запускается.Пример: если Задача A была запущена вчера и не выполнена, а теперь выполняется и сегодня, она не будет выполняться, если depends_on_past: True.

depen_on_past: если задано значение true, экземпляры задач будут запускаться последовательно, полагаясь на расписание предыдущей задачи.Экземпляр задачи для start_date разрешен к запуску.

Answer 2

Я не знаю, является ли это псевдокодом, но ваша зависимость должна выглядеть следующим образом (если вы говорите об операторах):

a >> b >> c

В графическом представлении строки появляются между -- б, а б --- в?Если зависимость не установлена ​​должным образом, вы увидите все эти три оператора просто «на графике» без линий, и поэтому они будут запланированы вместе.

Если вы хотите, чтобы каждая группа обеспечения доступности баз данных выполняла по одному, задайте

max_active_runs=1

в определении DAG() (не аргументы по умолчанию).

max_active_runs определяет количество допустимых одновременных экземпляров группы обеспечения доступности баз данных.

Наконец, вы можете сделать еще одну вещь - создать пул с 1 слотом и назначьте этот DAG этому пулу.