如何在 Airflow 中动态创建的任务之间创建依赖关系

Question

我正在使用以下代码创建动态任务。我想创建对这些动态创建的任务的依赖。例如，runStep_0 应该依赖于 runStep_1 等

for i in range(4):

 task = BashOperator(
     task_id='runstep_' + str(i),
     bash_command=cmd
     dag=dag)

Answer 1

终于找到出路了。在每次迭代期间将每个任务添加到列表中，并从列表中引用它。

Answer 2

在第一次迭代的循环内，将当前任务保存到 previous_task 变量。

第一次迭代后，只需设置 task.set_upstream(previous_task) 并使用 previous_task = task.

更新变量

Answer 3

类似的问答是here。将任务添加到列表中，然后添加一个简单的一行来绑定每个任务之间的依赖关系

a = []
for i in range(0,10):
    a.append(DummyOperator(
        task_id='Component'+str(i),
        dag=dag))
    if i not in [0]: 
        a[i-1] >> a[i]

生成此图

Answer 4

设置一个 previous_task 变量，如 Jorge 提到的，在我看来是最易读的解决方案，特别是如果每次迭代有多个任务时。

这里是代码：

previous_task = None
for i in range(4):
    
    task = BashOperator(
        task_id=f'task-{i}',
        bash_command=f'echo {i}',
    )

    if previous_task is not None:
       previous_task >> task
    else:
       previous_task = task

这里是多任务的例子

previous_task = None
for i in range(4):
    
    task_x = BashOperator(
        task_id=f'task_x-{i}',
        bash_command=f'echo {i}',
    )

    task_y = BashOperator(
        task_id=f'task_y-{i}',
        bash_command=f'echo {i}',
    )
    task_x >> task_y

    if previous_task is not None:
       previous_task >> task_x
    else:
       previous_task = task_y

如何在 Airflow 中动态创建的任务之间创建依赖关系

How to create dependency between dynamically created tasks in Airflow

airflow