Airflow ExternalTaskSensor触发DAG完成后仍无限等待的问题排查

1. 为什么触发的DAG成功后传感器仍无限等待？ #

核心原因是传感器查询时用的execution_date和jobs_dag实际运行任务的execution_date不匹配。你日志显示传感器在查2小时前的DAG，说明它拿着这个时间去搜jobs_dag的成功任务实例，但jobs_dag实际执行的任务对应的execution_date和这个时间对不上，自然找不到匹配项，就会一直等待。

2. 是否与execution_date或execution_delta默认值有关？ #

肯定有关：

• execution_date默认逻辑：ExternalTaskSensor默认用自身所在DAG的execution_date作为查询条件。如果你的trigger_dag是按调度周期跑（比如每天凌晨执行），而触发jobs_dag时用的是触发时刻的时间作为jobs_dag的execution_date（TriggerDagRunOperator默认行为），两者的execution_date就会不一致，传感器查不到对应任务。
• execution_delta默认值：如果没设置这个参数，它默认是None，但如果不小心配置了execution_delta=timedelta(hours=2)，传感器就会用当前DAG的execution_date减去2小时作为查询时间，这就会出现你日志里查2小时前DAG的情况。另外还要注意时区问题——如果Airflow核心时区是UTC，而你本地用UTC+2，你看到的jobs_dag执行时间是本地时间，和传感器用UTC时间查询的结果差2小时，也会导致匹配失败。

3. 如何正确同步两个DAG？ #

根据你的场景，推荐两种精准匹配的方案：

方案一：触发时传递execution_date，传感器直接复用

在trigger_dag里用TriggerDagRunOperator触发jobs_dag时，明确把trigger_dag的execution_date传给jobs_dag：

from airflow.operators.trigger_dagrun import TriggerDagRunOperator

trigger_jobs = TriggerDagRunOperator(
    task_id="trigger_jobs_dag",
    trigger_dag_id="jobs_dag",
    execution_date="{{ execution_date }}",  # 把当前DAG的execution_date传给jobs_dag
    dag=dag
)

然后在trigger_dag的ExternalTaskSensor里，直接用相同的execution_date查询：

from airflow.sensors.external_task import ExternalTaskSensor

wait_for_jobs = ExternalTaskSensor(
    task_id="wait_for_jobs_dag",
    external_dag_id="jobs_dag",
    external_task_id="jobs_dag里要等待的任务ID",  # 比如jobs_dag的最后一个任务ID
    execution_date="{{ execution_date }}",  # 和触发时的execution_date保持一致
    mode="reschedule",  # 用reschedule模式，避免长时间占用worker资源
    poke_interval=60,  # 每60秒检查一次
    dag=dag
)

方案二：通过XCom传递execution_date（适合复杂场景）

如果触发逻辑更复杂，比如jobs_dag的execution_date不是直接复用trigger_dag的，可以在jobs_dag的最后一个任务把自身的execution_date推送到XCom：

from airflow.operators.python import PythonOperator

def push_exec_date(**context):
    context["ti"].xcom_push(key="jobs_exec_date", value=context["execution_date"])

push_xcom_task = PythonOperator(
    task_id="push_exec_date",
    python_callable=push_exec_date,
    provide_context=True,
    dag=jobs_dag
)

然后在trigger_dag的传感器里，自定义execution_date_fn来拉取这个XCom（或者查询jobs_dag的成功DAG Run）：

from airflow.sensors.external_task import ExternalTaskSensor
from airflow.models import DagRun

def get_jobs_exec_date(**context):
    # 查询jobs_dag最新的成功运行实例的execution_date
    success_runs = DagRun.find(dag_id="jobs_dag", state="success")
    if success_runs:
        return max(run.execution_date for run in success_runs)
    return None

wait_for_jobs = ExternalTaskSensor(
    task_id="wait_for_jobs_dag",
    external_dag_id="jobs_dag",
    external_task_id="push_exec_date",
    execution_date_fn=get_jobs_exec_date,
    mode="reschedule",
    poke_interval=60,
    dag=dag
)

另外要注意：如果是时区问题导致的时间差，先统一Airflow的时区配置（修改airflow.cfg里的default_timezone），确保所有DAG的execution_date都用同一个时区计算。

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