在通过 for 循环动态生成任务后执行单个任务

Question

假设我有以下DAG(基本占位符函数(，它使用for循环来动态生成任务(通过迭代列表(：

from airflow import DAG
from airflow.operators.python_operator import PythonOperator
default_args = {
'owner': 'ETLUSER',
'depends_on_past': False,
'start_date': datetime(2019, 12, 16, 0, 0, 0),
'email': ['[e-mail]'],
'email_on_failure': False,
'email_on_retry': False,
'retries': 1,
'retry_delay': timedelta(minutes=5)
}
dag = DAG('xxx', catchup=False,
default_args=default_args, schedule_interval='0 */4 * * *')

# Some dummy function
def StepOne(x):
print(x)

def StepTwo():
print("Okay, we finished all of Step 1.")

some_list = [1, 2, 3, 4, 5, 6]

for t in some_list:
task_id = f'FirstStep_{t}'
task = PythonOperator(
task_id=task_id,
python_callable=StepOne,
provide_context=False,
op_kwargs={'x': str(t)},
dag=dag
)
task

我想介绍一些额外的任务，很简单：

task2 = PythonOperator(
task_id="SecondStep",
python_callable=StepTwo,
provide_context=False,
dag=dag
)

只有在第一个步骤中的所有步骤完成后才会运行。从表面上看，这将是task >> task2

我该怎么做？

Answer 1

你可以对数组有任务依赖关系。

在任务 A 和任务 B 完成后执行任务 C。

[taskA, taskB] >> taskC

或

在任务 A 完成后并行执行任务 B 和任务 C。

taskA >> [taskB, taskC]

只要上游或下游的 1 端是非阵列的。

因此，对于您的示例，

task1 = []
for t in some_list:
task_id = f'FirstStep_{t}'
task1.append(PythonOperator(
task_id=task_id,
python_callable=StepOne,
provide_context=False,
op_kwargs={'x': str(t)},
dag=dag))
task2 = PythonOperator(
task_id="SecondStep",
python_callable=StepTwo,
provide_context=False,
dag=dag)
task1 >> task2