Airflow任务依赖能否复用？复用实现方法求解

你的问题核心在于：Airflow的任务依赖是任务对象之间的关系赋值，不是可返回的数值，所以单纯执行t1 >> t2的函数不会自动把依赖关联到目标DAG里。下面给你几种直接可用的解决办法：

方法1：封装返回依赖链末尾任务的函数 #

把需要复用的任务和依赖逻辑封装成函数，返回依赖链的最后一个任务，这样后续可以直接用这个返回的任务继续拼接新的依赖。

示例代码：

from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime

# 封装DAG1的任务和依赖，返回末尾任务T2
def build_dag1_dependencies(dag):
    t1 = BashOperator(
        task_id="t1",
        bash_command="echo '执行任务T1'",
        dag=dag
    )
    t2 = BashOperator(
        task_id="t2",
        bash_command="echo '执行任务T2'",
        dag=dag
    )
    # 建立依赖
    t1 >> t2
    # 返回链的最后一个任务，方便后续拼接
    return t2

# 构建DAG2：复用DAG1的依赖，再加T3
with DAG(
    dag_id="dag_2",
    start_date=datetime(2024, 1, 1),
    schedule_interval="@daily"
) as dag2:
    # 获取DAG1依赖链的末尾任务T2
    dag1_last_task = build_dag1_dependencies(dag2)
    # 添加新任务T3并关联依赖
    t3 = BashOperator(
        task_id="t3",
        bash_command="echo '执行任务T3'",
        dag=dag2
    )
    dag1_last_task >> t3

# 构建DAG3：复用前面的依赖链，继续扩展
with DAG(
    dag_id="dag_3",
    start_date=datetime(2024, 1, 1),
    schedule_interval="@daily"
) as dag3:
    # 先拿到DAG1依赖链的T2
    dag1_last_task = build_dag1_dependencies(dag3)
    # 添加T3
    t3 = BashOperator(
        task_id="t3",
        bash_command="echo '执行任务T3'",
        dag=dag3
    )
    dag1_last_task >> t3
    # 添加T4-T6并行任务
    t4 = BashOperator(task_id="t4", bash_command="echo T4", dag=dag3)
    t5 = BashOperator(task_id="t5", bash_command="echo T5", dag=dag3)
    t6 = BashOperator(task_id="t6", bash_command="echo T6", dag=dag3)
    t3 >> [t4, t5, t6]
    # 添加T7
    t7 = BashOperator(task_id="t7", bash_command="echo T7", dag=dag3)
    [t4, t5, t6] >> t7

方法2：封装仅设置依赖的函数（适用于任务已单独定义的场景） #

如果T1、T2这类任务需要在不同DAG里单独实例化（比如参数不同），可以封装一个只负责建立依赖关系的函数，直接操作传入的任务对象：

def setup_dag1_deps(t1, t2):
    # 直接给传入的任务建立依赖
    t1 >> t2

# 在DAG中使用：
with DAG("dag_2", ...) as dag:
    t1 = BashOperator(task_id="t1", bash_command="echo T1", dag=dag)
    t2 = BashOperator(task_id="t2", bash_command="echo T2", dag=dag)
    # 调用函数建立依赖
    setup_dag1_deps(t1, t2)
    t3 = BashOperator(task_id="t3", ...)
    t2 >> t3

方法3：用TaskGroup封装复用整个任务组 #

如果需要复用的是一整套任务（不仅仅是依赖），推荐用Airflow的TaskGroup把相关任务和依赖打包成一个组，这样可以直接在其他DAG中引用整个组：

from airflow.utils.task_group import TaskGroup

def get_dag1_task_group(dag):
    with TaskGroup("dag1_task_group", dag=dag) as tg:
        t1 = BashOperator(task_id="t1", bash_command="echo T1")
        t2 = BashOperator(task_id="t2", bash_command="echo T2")
        t1 >> t2
    return tg

# 在DAG2中引用：
with DAG("dag_2", ...) as dag:
    dag1_tg = get_dag1_task_group(dag)
    t3 = BashOperator(task_id="t3", ...)
    # 整个任务组作为一个节点，直接和T3建立依赖
    dag1_tg >> t3

关键说明 #

之前的函数失效，本质是因为你只在函数里执行了t1 >> t2，但没有把这些任务关联到目标DAG，也没有提供后续可拼接的入口。上面的三种方法要么返回可继续链接的任务/任务组，要么直接操作传入的任务对象建立依赖，完美解决复用问题。

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