2

假设我有以下 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': ['xxx@xxx.com'],
    '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

我该怎么做呢？

4

1 回答 1

3

您可以与数组有任务依赖关系。

在 taskA 和 taskB 都完成后执行 taskC。

[taskA, taskB] >> taskC

或者

在 taskA 完成后并行执行 taskB 和 taskC。

taskA >> [taskB, taskC]

只要上游或下游的一侧是非阵列的。

因此，对于您的示例，

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
于 2020-01-09T19:54:11.610 回答