2

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

for i in range(4):

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

4 回答 4

3

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

于 2016-10-11T09:04:22.490 回答
2

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

在第一次迭代之后，只需设置 task.set_upstream(previous_task) 并使用 previous_task = task 更新变量。

于 2017-05-07T18:04:54.223 回答
0

在我看来，像 Jorge 提到的那样设置 previous_task 变量是最易读的解决方案，尤其是在每次迭代有多个任务的情况下。

这里的代码：

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
于 2021-02-25T22:37:12.487 回答
0

一个类似的问题和答案是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]

生成此图 输出

于 2020-11-23T00:05:50.377 回答