如何在Airflow中正确将子链首尾接入主链？

你遇到的核心问题是Airflow中>>运算符会返回链中的最后一个Operator，所以用(A >> B >> C)赋值的变量实际指向C，导致分支节点错误地连接到子链末尾而非起始节点。

以下是几种无需逐个为中间Operator赋值变量的简洁写法：

方案1：使用>>=保留起始节点引用 #

利用Airflow的__irshift__（>>=）运算符，它会返回起始Operator本身，而非链的末尾节点：

# 定义子链，sod始终指向起始节点
sod = DummyOperator(task_id="sod")
sod >>= DummyOperator(task_id="sod_do_this") >> DummyOperator(task_id="sod_last")

# 获取子链末尾节点用于连接end
sod_last = sod.get_downstream_list()[-1]

# 构建主链
should_run_sod = BranchPythonOperator(
    task_id="should_run_sod",
    python_callable=lambda: "sod", # 返回起始节点的task_id
)
no_sod = DummyOperator(task_id="no_sod")
end = DummyOperator(task_id="end")

should_run_sod >> [sod, no_sod]
sod_last >> end
no_sod >> end

方案2：拆分链式赋值，同时保留起始和末尾节点 #

这种写法更直观，仅为起始和末尾节点赋值，中间节点无需单独命名：

# 起始节点
sod = DummyOperator(task_id="sod")
# 链式连接，sod_last指向子链末尾
sod_last = sod >> DummyOperator(task_id="sod_do_this") >> DummyOperator(task_id="sod_last")

# 主链构建
should_run_sod = BranchPythonOperator(
    task_id="should_run_sod",
    python_callable=lambda: "sod",
)
no_sod = DummyOperator(task_id="no_sod")
end = DummyOperator(task_id="end")

should_run_sod >> [sod, no_sod]
sod_last >> end
no_sod >> end

方案3：使用TaskGroup（Airflow 2.0+推荐） #

TaskGroup可以将子链封装为一个逻辑单元，同时自动管理任务ID的命名空间，写法更整洁：

from airflow.utils.task_group import TaskGroup

# 封装子链到TaskGroup
with TaskGroup("sod_subchain") as sod_group:
    # 内部链式定义，无需单独赋值
    DummyOperator(task_id="sod") >> DummyOperator(task_id="sod_do_this") >> DummyOperator(task_id="sod_last")

# 主链构建
should_run_sod = BranchPythonOperator(
    task_id="should_run_sod",
    # 返回TaskGroup内起始任务的完整ID（group_id.task_id）
    python_callable=lambda: "sod_subchain.sod",
)
no_sod = DummyOperator(task_id="no_sod")
end = DummyOperator(task_id="end")

# 分支连接到整个TaskGroup，Airflow会自动关联到Group内的起始任务
should_run_sod >> [sod_group, no_sod] >> end

• 方案1和2适合简单子链，写法轻便；
• 方案3适合复杂子链，能更好地组织任务结构，尤其在多人协作或任务数量较多时更易维护。

内容的提问来源于stack exchange，提问作者KamilCuk