Блог

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

Что-то похожее на это:

Что я пытался запустить следующий код:

 dag = DAG(  'test',  default_args={"start_date": datetime(2019, 12, 5)},  schedule_interval=None )  start = DummyOperator(task_id='start', dag=dag)  end_opr = DummyOperator(task_id='end_opr', dag=dag)  dummy1 = DummyOperator(task_id='dummy', dag=dag)  dummy2 = DummyOperator(task_id='dummy2', dag=dag)  start gt;gt; [dummy1, dummy2] gt;gt; end_opr  

Но то, что я получаю, — это дубликат end_opr вместо dummy1 и dummy2 присоединиться к нему в конце.

Что я делаю не так?

Мой env: композитор-1.17.2-воздушный поток-1.10.15

То, что вы делаете, правильно! Однако вы сравниваете представление графика (первое изображение) с представлением дерева (второе изображение). В древовидном представлении отображается DAG для каждого отдельного пути от корня к листу. end_opr Задача на самом деле не дублируется, а скорее появляется дважды, потому что она является частью 2 различных путей. Проверьте представление графика в пользовательском интерфейсе для этой группы DAG; вы должны увидеть, к чему вы стремитесь там.