Как настроить двух рабочих в воздушном потоке

Question

У меня есть два рабочих и 3 задания.

dag = DAG('dummy_for_testing', default_args=default_args)

t1 = BashOperator(
    task_id='print_task1',
    bash_command='task1.py',
    dag=dag)

t2 = BashOperator(
    task_id='print_task2',
    bash_command='task2.py',
    dag=dag)

t3 = BashOperator(
    task_id='print_task3',
    bash_command='task3.py',
    dag=dag)

t1 >> t2 >> t3

Допустим, я выполняю tasks(t1,t2,t3) для определенного файла. В настоящее время все работает на одном работнике, но я хочу настроить другого работника, который будет принимать вывод первой задачи и выполнять задачу t2, а затем задачу t3. Таким образом, queue1 будет выполнять t1 для следующего файла. Как я могу сделать эту работу для двух работников. Я думаю об использовании queues, но не могу понять, как заставить queue2 ждать, пока задача t1 в queue1 завершится.

Answer 1

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

В приведенном вами примерезадачи могут выполняться полностью один worker 1, worker 2, или смесь обоих.Это потому, что t2 не начнется, пока t1 не завершится.В промежутке между t1 завершением и t2 запуском оба работника будут бездействовать (при условии, что у вас не работают другие дагы).Один из них выиграет гонку, зарезервировав задачу t2 для запуска.

Если вам нужно было выполнять конкретные задачи на разных рабочих (скажем, иметь одного или нескольких рабочих с доступными ресурсами более высокого уровня, или специальныеоборудование) вы можете указать очередь на уровне задач.Очередь не будет иметь значения в том порядке, в котором задачи выполняются, так как планировщик воздушного потока гарантирует, что задача не будет запущена до тех пор, пока задачи перед ней не будут успешно выполнены.