airflow - Airflow 的 BranchPythonOperator 如何工作？

Question

我正在努力理解 Airflow 中的 BranchPythonOperator 是如何工作的。我知道它主要用于分支，但对文档中关于传递到任务的内容以及我需要从任务上游传递/期望的内容感到困惑。

给出文档 on this page 中的简单示例名为 run_this_first 的上游任务和 2 个分支的下游任务的源代码是什么样的？ Airflow 究竟如何知道运行 branch_a 而不是 branch_b？上游任务的输出在哪里被注意到/读取？

Answer 1

您的 BranchPythonOperator 是使用 python_callable 创建的，它将是一个函数。该函数应根据您的业务逻辑返回您已连接的直接下游任务的任务名称。这可能是紧邻下游的 1 到 N 个任务。下游任务不需要读取任何内容，但是您可以使用 xcom 向它们传递元数据。

def decide_which_path():
    if something is True:
        return "branch_a"
    else:
        return "branch_b"


branch_task = BranchPythonOperator(
    task_id='run_this_first',
    python_callable=decide_which_path,
    trigger_rule="all_done",
    dag=dag)

branch_task.set_downstream(branch_a)
branch_task.set_downstream(branch_b)

设置 trigger_rule 非常重要，否则所有其余的都将被跳过，因为默认值为 all_success。