Apache Airflow跨DAG依赖：DAG A Task2成功后触发DAG B方案咨询

针对你描述的跨DAG依赖场景，以下是两种符合要求的可落地实现方案，均不需要在DAG A中新增额外执行时间规则： #

方案1：使用ExternalTaskSensor（无侵入DAG A的最优方案） #

这是Airflow原生提供的跨DAG依赖感知能力，完全不需要修改DAG A的现有代码，仅在DAG B侧配置依赖规则即可。
核心配置要点：

• 传感器会自动轮询DAG A中Task 2的运行状态，直到任务标记为成功后才会触发DAG B的后续任务
• 配置时需要对齐两个DAG的调度时间规则，适配Task 2无固定运行时长的特性
  示例代码：

from airflow.sensors.external_task import ExternalTaskSensor
from airflow import DAG
from datetime import datetime, timedelta

with DAG(
    dag_id="DAG_B",
    schedule_interval="0 17 * * *", # 和DAG A的调度时间保持一致即可，可按需调整
    start_date=datetime(2024,1,1),
    catchup=False
) as dag:
    # 定义依赖DAG A Task 2的状态感知传感器
    wait_for_dag_a_task2 = ExternalTaskSensor(
        task_id="wait_for_dag_a_task2_success",
        external_dag_id="DAG_A", # 替换为你的DAG A实际ID
        external_task_id="Task_2", # 替换为DAG A中Task 2的实际任务ID
        timeout=3600*12, # 最长轮询等待12小时，可根据实际业务场景调整
        poke_interval=60, # 每60秒轮询一次任务状态，频率可自定义
        mode="reschedule", # 该模式可避免长时间占用工作进程资源
    )

    # DAG B中原有的对应任务
    dag_b_target_task = ... # 保留你的原有业务逻辑即可

    # 定义依赖关系
    wait_for_dag_a_task2 >> dag_b_target_task

如果DAG B的调度时间和DAG A不一致，只需新增execution_delta参数匹配两者时间差即可，比如DAG B是每日17:30调度，添加配置execution_delta=timedelta(minutes=30)。

方案2：使用TriggerDagRunOperator（即时触发无轮询方案） #

如果允许在DAG A中新增一个轻量触发任务（无任何额外时间规则，仅做触发动作），可以选择这个方案，触发即时性更高，不需要轮询占用资源。
仅需要在DAG A原有依赖关系的基础上新增一个触发任务即可，不需要修改DAG A原有调度配置和业务逻辑：

# 仅在DAG A中新增以下触发任务，不改动原有调度和业务逻辑
from airflow.operators.trigger_dagrun import TriggerDagRunOperator

trigger_dag_b_task = TriggerDagRunOperator(
    task_id="trigger_dag_b",
    trigger_dag_id="DAG_B", # 替换为DAG B的实际ID
    wait_for_completion=False, # 不需要等待DAG B运行完成，不影响DAG A原有Task 3的执行
)

# 调整DAG A原有依赖关系即可
Task_1 >> Task_2 >> [Task_3, trigger_dag_b_task]

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