在通过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': ['[email protected]'],
    '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
  )

仅在第一步中的all完成后运行。线性地，这将是task >> task2

我该怎么做？

Answer 1

您可以将数组作为依赖项

[taskA, taskB] >> taskC