气流分支操作器链

Question

我正在尝试使用气流分支的操作器。我得到了一个错误：

def next_task(x):
  if x : return "branch_data"
  else: return "branch_no_data"

branch_task  = BranchPythonOperator(
   task_id = 'branch_task',
   python_callable = next_task,
   op_args = [x],
   dag = dag
)

branch_no_data=DummyOperator(task_id="branch_no_data", dag=dag)
branch_data=DummyOperator(task_id="branch_data", dag=dag)
join_task=DummyOperator(task_id="join_task", dag=dag)
branch_task >> [branch_data, branch_no_data ]  >>  join_task

我只需要执行两个任务中的一个（分支数据、分支无数据）。 下面是我实现任务链接的失败尝试。 我得到了一个错误：

branch_task >> [branch_data, branch_no_data ]  >>  join_task
TypeError: unsupported operand type(s) for >>: 'BranchPythonOperator' and 'list'

Answer 1

气流不允许将列表链接到其他任务。 一个任务可以输出到任务列表中，但任务列表不能链接到单个任务中。因此list类型不支持>>操作数的错误

为了您自己的清晰性和避免任何潜在的问题，最好只单独列出链，而不列出列表

您的代码有两个版本可以使用：

branch_task >> branch_data    >> join_task
branch_task >> branch_no_data >> join_task

或

branch_task    >> [ branch_data, branch_no_data ]
branch_data    >> join_task
branch_no_data >> join_task