Airflow：不同调度父子DAG使用ExternalTaskSensor未按预期工作咨询

定位到的代码核心问题 #

• ExternalTaskMarker 配置完全错误：父DAG内的Marker组件作用是关联子DAG的依赖链路，方便UI展示父子关系，你的代码里把external_dag_id写成了父DAG自身的ID，没有指向子DAG ID，完全没有起到关联作用
• ExternalTaskSensor 关键参数缺失/错配：
  • 未配置failed_states，当父任务执行失败时，Sensor只会因为没匹配到success状态而持续重试等待，直到达到超时时间才会报错，这就是你遇到的“父已经失败、子仍延迟报错”的核心原因
  • execution_delta 和双方DAG的实际调度规则不匹配，导致Sensor找不到对应执行时间的父任务，出现状态判断不准的问题
  • timeout=60 和你需求的“最长等待40分钟”完全不符，会导致提前超时失败
  • 跨DAG直接导入对方的DAG、任务对象的写法是Airflow 反模式，极易引发DAG解析冲突、循环依赖问题
• 父DAG基础导入缺失：代码里直接使用datetime但没有导入对应模块，本身就存在运行报错隐患
• 手动触发DAG时未匹配对应执行时间：ExternalTaskSensor是严格按照execution_date匹配任务状态的，手动触发父DAG时如果没有指定和子Sensor预期一致的执行时间，Sensor会完全找不到手动触发的父任务，直到超时才会报错

修正后的代码实现 #

父DAG（dag_etl1） #

import time
from datetime import datetime
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from airflow.sensors.external_task_sensor import ExternalTaskMarker

start_date = datetime(2021, 3, 1, 20, 36, 0)

class Exept(Exception):
    pass

def wait():
    time.sleep(3)
    with open('etl.txt', 'r') as txt:
        line = txt.readline()
        if line == 'err':
            print(1)
            raise Exept
    return 'etl success'

with DAG(
    dag_id="dag_etl1",
    start_date=start_date,
    # 生产环境替换为 30 * * * *，测试可临时用 * * * * *
    schedule_interval='* * * * *',
    tags=['example2'],
) as etl1:
    wait_timer = PythonOperator(task_id='wait_timer', python_callable=wait)
    # 修正ExternalTaskMarker配置：指向子DAG的对应Sensor任务
    parent_task = ExternalTaskMarker(
        task_id="parent_task",
        external_dag_id="etl_child",
        external_task_id="wait_for_parent_etl1",
    )
    wait_timer >> parent_task

子DAG（etl_child） #

from datetime import datetime
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from airflow.sensors.external_task_sensor import ExternalTaskSensor

start_date = datetime(2021, 3, 1, 20, 36, 0)

def check():
    return 'I succeeded'

# 灵活计算对应父任务的执行时间，避免固定delta的适配问题
def get_parent_exec_date(child_exec_date):
    # 业务逻辑说明：子DAG在X:01触发，等待同小时X:30的父任务完成
    # 如果业务是等上一个小时的父任务，调整这里的时间逻辑即可
    return child_exec_date.replace(minute=30, second=0, microsecond=0)

with DAG(
    dag_id='etl_child', 
    start_date=start_date, 
    # 生产环境替换为 1 8-17 * * 1-5，测试可临时用 * * * * *
    schedule_interval='* * * * *',
    tags=['testing_child_dag']
) as etl_child:
    wait_for_parent = ExternalTaskSensor(
        task_id="wait_for_parent_etl1",
        # 直接用字符串指定外部DAG、任务ID，不做跨DAG导入
        external_dag_id="dag_etl1",
        external_task_id="parent_task",
        allowed_states=['success'],
        # 配置失败状态：父任务出现以下状态时，Sensor直接失败终止，不等待
        failed_states=['failed', 'upstream_failed', 'skipped'],
        mode='reschedule',
        # 用自定义函数匹配父任务执行时间，替代容易出错的固定delta
        execution_date_fn=get_parent_exec_date,
        # 40分钟 = 2400秒，匹配需求的最长等待时长
        timeout=2400,
    )

    task1 = PythonOperator(task_id='task1', python_callable=check)
    wait_for_parent >> task1

运行注意事项 #

• 手动触发DAG时，必须在触发弹窗的「Execution Date」项选择与业务逻辑对应的时间，确保父子任务的execution_date能被Sensor正确匹配，否则Sensor会找不到目标父任务
• 测试过程中如果临时调整DAG的调度间隔，一定要同步修改get_parent_exec_date函数里的时间匹配逻辑，保证执行时间的对应关系正确
• ExternalTaskMarker仅用于UI层面的依赖链路展示，不承担状态同步逻辑，所有的等待、状态判断逻辑全部由ExternalTaskSensor实现

内容的提问来源于stack exchange，提问作者Дмитрий Пырин