如何将一个DAG的任务成功状态引用至另一个DAG

这个需求完全可以实现，以下以最通用的Apache Airflow调度场景为例，给出可直接落地的操作方案，完全适配你提到的「两DAG同调度日期、不同计划运行时间、仅依赖DAG A中单个任务成功状态」的要求。

具体实现方案 #

方案1：用原生ExternalTaskSensor实现（优先推荐，无额外依赖） #

这是Airflow原生提供的跨DAG任务状态探测组件，配置简单、稳定性高，是这类场景的标准解法：

• 首先确认基础配置对齐
  你当前两个DAG的逻辑调度日期完全一致，只是运行时间不同，天然满足时间匹配要求，只要保证两个DAG的调度时区配置一致、没有自定义偏移逻辑日期的配置即可。
• 在DAG B中新增探测任务，作为Task3的直接上游
  参考代码如下，替换成你自己的DAG、任务ID即可直接用：

  from airflow.sensors.external_task import ExternalTaskSensor
  
  # 探测任务，放在DAG B的定义代码中
  wait_for_task1 = ExternalTaskSensor(
      task_id='wait_for_dag_a_task1',
      external_dag_id='dag_a', # 替换为DAG A实际配置的dag_id
      external_task_id='task_1', # 替换为DAG A中task1实际配置的task_id
      allowed_states=['success'], # 仅task1为成功状态时才放行
      failed_states=['failed', 'skipped'], # task1失败/跳过时直接标记探测失败，不无限阻塞
      execution_date_fn=lambda exec_date: exec_date, # 两DAG逻辑日期一致，直接用当前执行日期匹配即可
      poke_interval=60, # 每60秒查一次task1的状态，不用太频繁浪费资源
      timeout=21600, # 超时时间设为6小时，覆盖早8点到下午2点的时间差，留足冗余避免DAG A偶发延迟导致误超时
      mode='reschedule' # 探测未通过时释放worker资源，不会长期占着计算slot
  )
  
  # 配置依赖关系：探测通过后才允许跑Task3
  wait_for_task1 >> task3
  

• 运行逻辑验证
  哪怕到了DAG B下午2点的计划启动时间，只要探测任务没查到同调度日期下DAG A的task1成功状态，Task3就会一直等待不会执行；如果DAG A的task1在下午2点前就跑成功了，到了DAG B的调度时间会直接放行Task3，完全符合需求。

方案2：自定义状态标记（适配低版本Airflow或定制化场景） #

如果你用的Airflow版本过低，ExternalTaskSensor功能不稳定，可以用变量标记的方式实现：

• 在DAG A的task1上配置成功回调，task1跑成功后，往Airflow元数据库写一个带当前执行日期标识的变量，比如标记dag_a_task1_{{ execution_date }}_success = True
• 在DAG B的Task3最开头加一段判断逻辑，读取对应执行日期的标记变量，只有变量值为True时才继续执行业务逻辑，否则抛出重试异常，等下次调度周期再判断
• 记得加个后置清理逻辑，每天跑完后把历史日期的标记变量删掉，避免元数据库存太多冗余数据。

注意事项 #

• Airflow不支持直接在不同DAG的任务间写>>配置跨DAG依赖，必须通过上述探测或者状态标记的方式实现
• 如果两个DAG部署在不同的Airflow集群，把ExternalTaskSensor的探测逻辑换成调用DAG A所在集群的任务状态查询接口即可，核心逻辑不变
• 不要把poke_interval设的太短，10-60秒间隔足够，太频繁会给元数据库造成不必要的压力

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