Airflow 本身不支持直接复用 >> 依赖表达式，但可通过封装任务创建逻辑并返回链中最后一个 Operator 实现高效复用，从而避免重复代码、提升 DAG 可维护性。
airflow 本身不支持直接复用 `>>` 依赖表达式，但可通过封装任务创建逻辑并返回链中最后一个 operator 实现高效复用，从而避免重复代码、提升 dag 可维护性。
在 Airflow 中，任务依赖（如 T1 >> T2）是副作用操作，而非可赋值或传递的值，因此无法像普通对象一样被“返回”或“拼接”。但实践中，我们真正需要复用的往往是一组固定结构的任务及其拓扑关系——例如始终以 t1 → t2 开头的流水线。此时，核心思路是：将任务定义与依赖设置封装为函数，并显式返回链末端的 Operator，作为后续依赖扩展的接入点。
以下是一个规范、可复用的实现模式：
from datetime import datetime
from airflow import DAG
from airflow.operators.empty import EmptyOperator
def generate_t1_t2() -> EmptyOperator:
"""
创建并连接 t1 和 t2，返回 t2 作为依赖链出口。
此函数可在任意 DAG 中调用，确保 t1→t2 结构完全一致。
"""
t1 = EmptyOperator(task_id="t1")
t2 = EmptyOperator(task_id="t2")
t1 >> t2
return t2
# DAG 1：仅使用基础链
with DAG("dag1", start_date=datetime(2025, 1, 1), schedule=None) as dag1:
generate_t1_t2()
# DAG 2：在基础链后追加单个任务
with DAG("dag2", start_date=datetime(2025, 1, 1), schedule=None) as dag2:
t2 = generate_t1_t2()
t3 = EmptyOperator(task_id="t3")
t2 >> t3
# DAG 3：支持复杂分支与汇聚
with DAG("dag3", start_date=datetime(2025, 1, 1), schedule=None) as dag3:
t2 = generate_t1_t2()
t3 = EmptyOperator(task_id="t3")
t4, t5, t6 = [EmptyOperator(task_id=f"t{i}") for i in (4, 5, 6)]
t7 = EmptyOperator(task_id="t7")
t2 >> t3
t3 >> [t4, t5, t6]
[t4, t5, t6] >> t7
✅ 关键设计要点：
- 函数必须返回链中最后一个 Operator（如
t2），它是后续>>操作的合法左操作数； - 所有任务应在函数内完整定义并建立依赖，避免跨作用域引用未初始化对象；
- 若需复用多段子链（如
t1→t2和t4→t5→t6），可为每段编写独立生成函数，再按需组合。
⚠️ 注意事项：
- 不要尝试在函数中返回
t1 >> t2表达式本身（它返回None）； - 避免在函数外提前声明任务变量（如
t1 = ...放在函数外），否则会破坏 DAG 隔离性，导致任务被多个 DAG 共享而引发运行时错误； - 若子链需参数化（如动态
task_id或python_callable），应将参数加入函数签名，增强灵活性。
这种模式本质是将“依赖模板”转化为可组合的构建器函数，既符合 Airflow 的执行模型，又显著提升了大型项目中 DAG 的一致性与可维护性——尤其适用于标准化数据预处理、ETL 初始化、监控上报等通用前置流程。
