Airflow中ExternalTaskSensor未正常工作问题排查求助

核心原因分析 #

你的问题本质是ExternalTaskSensor默认匹配的执行日期与已完成的外部DAG实例不匹配，导致传感器找不到符合条件的任务状态，从而持续等待。

1. 执行日期不匹配 #

ExternalTaskSensor默认会等待与当前DAG运行实例执行日期完全相同的外部DAG任务：

• 你的test_first_dag起始日期是2022-01-01，test_second_dag起始日期是2022-01-02
• 触发test_second_dag时，它的执行日期为2022-01-02（或手动触发的日期），传感器会去查找test_first_dag中执行日期为2022-01-02的task1
• 但你之前执行的test_first_dag实例执行日期是2022-01-01，两者不匹配，传感器自然找不到已完成的任务

2. 其他可能的次要原因 #

• 外部任务状态异常：如果test_first_dag的task1最终状态不是success（比如skipped或failed），传感器也会持续等待
• 传感器轮询模式问题：默认poke模式会占用worker资源持续轮询，看起来像是一直卡住，但实际是在重复检查

解决方案 #

方法1：指定执行日期偏移量 #

通过execution_delta参数，让传感器等待当前执行日期之前特定时间间隔的外部DAG实例。修改test_second_dag中的传感器配置：

wait_for_dag1_task1 = ExternalTaskSensor(
    task_id='wait_for_dag1_task1',
    external_dag_id='test_first_dag',
    external_task_id='task1',
    # 让传感器等待当前执行日期前1天的外部DAG实例
    execution_delta=timedelta(days=1),
    # 改为reschedule模式，轮询间隔期间释放worker资源（更推荐）
    mode='reschedule',
    # 设置轮询间隔，默认是60秒，可根据需求调整
    poke_interval=60,
    dag=dag,
)

方法2：自定义执行日期匹配逻辑 #

如果需要更灵活的日期匹配，使用execution_date_fn自定义函数来指定要等待的外部任务执行日期：

def get_external_execution_date(context):
    # 返回当前DAG执行日期前1天的日期
    return context['execution_date'] - timedelta(days=1)

wait_for_dag1_task1 = ExternalTaskSensor(
    task_id='wait_for_dag1_task1',
    external_dag_id='test_first_dag',
    external_task_id='task1',
    execution_date_fn=get_external_execution_date,
    mode='reschedule',
    dag=dag,
)

额外检查步骤 #

1. 确认test_first_dag的task1最终状态是success：在Airflow UI中查看该任务的运行日志和状态
2. 验证执行日期匹配：触发test_second_dag后，查看传感器的日志，确认它在查找的外部任务执行日期是否正确

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