Airflow 2.8中ExternalTaskSensor实现DAG依赖失败求助

问题原因 #

你遇到的超时问题核心在于：ExternalTaskSensor默认会匹配与当前任务拥有相同execution_date的外部DAG任务。当你先手动触发dag2，再触发dag1时，两者的execution_date是不同的触发时间，导致sensor无法找到对应的dag1任务实例，最终在60秒超时后失败。

解决方案 #

要实现"先启动dag2，等待dag1完成后再执行task2"的逻辑，需要让sensor忽略execution_date的匹配，改为跟踪dag1的最新成功运行实例。以下是两种可行的修改方案：

方案1：使用execution_date_fn匹配最新成功的dag1实例 #

修改dag2中的ExternalTaskSensor配置，通过execution_date_fn指定获取dag1最新成功execution_date的逻辑：

from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.sensors.external_task import ExternalTaskSensor
from airflow.utils.db import provide_session
from airflow.models import DagRun
from datetime import datetime

default_args = {
    'start_date': datetime(2025, 2, 10),
}

@provide_session
def get_latest_dag1_success_date(session=None, **context):
    # 查询dag1最新的成功运行实例
    dag_run = session.query(DagRun)\
        .filter(DagRun.dag_id == 'dag1', DagRun.state == 'success')\
        .order_by(DagRun.execution_date.desc())\
        .first()
    return dag_run.execution_date if dag_run else None

with DAG('dag2', default_args=default_args, schedule_interval=None, catchup=False, tags=['infa']) as dag2:
    wait_for_dag1 = ExternalTaskSensor(
        task_id='wait_for_dag1',
        external_dag_id='dag1',
        external_task_id='dag1_complete_marker',
        mode='poke',
        timeout=3600,  # 延长超时时间，给dag1足够运行时长
        poke_interval=10,
        execution_date_fn=get_latest_dag1_success_date,
        allowed_states=['success'],
        failed_states=['failed', 'up_for_retry'],
    )

    task2 = BashOperator(
        task_id='task2',
        bash_command='echo "This is dag2"',
    )

    wait_for_dag1 >> task2

方案2：等待整个dag1完成（无需指定单个任务） #

如果只需要等待dag1整体完成，无需关注特定任务，可以省略external_task_id，同时保留核心匹配逻辑：

wait_for_dag1 = ExternalTaskSensor(
    task_id='wait_for_dag1',
    external_dag_id='dag1',
    external_task_id=None,  # 不指定则等待整个dag1完成
    mode='poke',
    timeout=3600,
    poke_interval=10,
    execution_date_fn=get_latest_dag1_success_date,
    allowed_states=['success'],
    failed_states=['failed', 'up_for_retry'],
)

额外注意事项 #

• 务必为DAG设置明确的start_date，避免因默认值导致的execution_date异常。
• 根据dag1的实际运行时长调整timeout参数，60秒通常不足以覆盖常规任务的执行时间。
• 可根据需求调整poke_interval，平衡检查频率与数据库查询开销。

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