Airflow如何处理同调度周期跨DAG执行状态依赖问题

DAG A与DAG B配置为同一调度时间触发（例如每日上午10点）：DAG A通常5分钟即可执行完成，DAG B需等待校验DAG A的执行状态，若DAG A执行成功则进入后续步骤，否则抛出错误。

DAG B仅可读取同一执行日期/时间对应的DAG A运行实例状态，不得复用历史运行结果：例如DAG A昨日运行成功但当日因故障未启动，已启动的DAG B不得读取昨日的成功状态，必须校验当日对应DAG A实例的实时状态。

• 如何实现上述同执行时间的跨DAG状态匹配逻辑，避免误读取历史DAG运行记录
• 当DAG A实例状态处于成功、失败之外的其他状态时，代码层面应如何处理

你之前写的查询逻辑存在明显漏洞：按执行时间倒序取最新1条记录的写法，既没有过滤目标DAG ID，也没有绑定当前DAG的执行时间，极容易读取到其他DAG或者历史日期的运行记录，完全不符合规则要求。下面是可直接落地的实现方式：

1. 同执行时间跨DAG状态的精确匹配逻辑 #

Airflow中同一调度时间触发的两个DAG，生成的DagRun实例的execution_date是完全时区对齐、值相等的，这是匹配同批次DAG实例的唯一可靠键，绝对不要用「取最新记录」的逻辑查询。

优先用官方原生组件实现（推荐） #

不要自己手写数据库查询，直接用Airflow内置的ExternalTaskSensor即可，只要不手动配置执行时间偏移，组件默认就会严格匹配和当前DAG实例相同execution_date的外部DAG实例，从根源上避免读取历史记录，示例配置：

from datetime import timedelta
from airflow import DAG
from airflow.sensors.external_task import ExternalTaskSensor
from airflow.utils.state import DagRunState

# 假设dag_b是你已经定义好的DAG B实例
wait_for_dag_a = ExternalTaskSensor(
    task_id="wait_for_dag_a_success",
    external_dag_id="dag_a", # 替换为你的DAG A实际ID
    # 若要等待整个DAG A执行完成，填写DAG A最后一个任务的ID即可
    external_task_id="dag_a_final_task",
    # 核心配置：执行时间差为0，严格匹配同调度批次的实例，不会读历史数据
    execution_delta=timedelta(0),
    poke_interval=30, # 每30秒轮询一次状态
    timeout=3600, # 最长等待1小时，超时自动失败
    failed_states=[DagRunState.FAILED, DagRunState.SHUTDOWN, DagRunState.REMOVED],
    mode="reschedule", # 等待时释放worker资源，不占用执行slot
    dag=dag_b
)

自定义查询的正确写法 #

如果你确实需要自己写查询逻辑，必须加两个精确过滤条件：dag_id等于DAG A的ID、execution_date等于当前上下文拿到的执行时间，示例代码：

from airflow.models import DagRun
from airflow.exceptions import AirflowException
from airflow.utils.state import DagRunState
from airflow import settings

def check_dag_a_status(**context):
    current_exec_date = context["execution_date"]
    target_dag_id = "dag_a"

    # 精确查询匹配的DAG A实例
    dag_a_instance = settings.Session().query(DagRun).filter(
        DagRun.dag_id == target_dag_id,
        DagRun.execution_date == current_exec_date
    ).one_or_none()

    # 对应实例不存在，说明DAG A当日未启动，直接报错
    if not dag_a_instance:
        raise AirflowException(f"执行日期为{current_exec_date}的DAG A实例不存在，校验失败")
    
    return dag_a_instance.state

注意：禁止使用order_by(DR.execution_date.desc()).limit(1)的写法，这是误读历史数据的核心原因。

2. 非成功/失败状态的处理规则 #

DAG实例除了success、failed之外，还存在多类中间状态和异常终止状态，处理逻辑如下：

• 中间状态（queued、running、up_for_retry、up_for_reschedule、deferred）：既不判定成功也不判定失败，进入轮询等待，间隔指定时间后重新查询状态
• 异常终止状态（shutdown、removed）：直接抛出异常，终止DAG B流程
• 必须配置合理的超时时间，避免无限轮询占用资源

如果是自己实现轮询逻辑，可以参考下面的状态判断分支：

# 等待状态集合：这些状态下继续轮询
WAIT_STATES = {
    DagRunState.QUEUED,
    DagRunState.RUNNING,
    DagRunState.UP_FOR_RETRY,
    DagRunState.UP_FOR_RESCHEDULE,
    DagRunState.DEFERRED
}
# 失败状态集合：这些状态下直接抛错
FAIL_STATES = {
    DagRunState.FAILED,
    DagRunState.SHUTDOWN,
    DagRunState.REMOVED
}

current_state = check_dag_a_status(**context)
if current_state == DagRunState.SUCCESS:
    # 校验通过，执行后续步骤
    proceed_to_next_step()
elif current_state in WAIT_STATES:
    # 等待下一次轮询
    return False
elif current_state in FAIL_STATES:
    raise AirflowException(f"DAG A实例状态为{current_state}，校验不通过，流程终止")

如果使用官方的ExternalTaskSensor，上述状态判断逻辑已经内置，只需要按需配置allowed_states和failed_states参数即可，不需要重复实现。

避坑提醒：不要为了兼容DAG A启动延迟的问题，放宽execution_date的匹配范围，否则会出现读取历史实例状态的问题。如果DAG A和DAG B的调度时间确实存在固定偏移，再用execution_delta配置固定的时间差即可。

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