Airflow导入ExternalTaskSensor异常及主DAG调度外部Dataflow作业方案咨询

关于ExternalTaskSensor的功能确认 #

明确支持该使用场景：

• 你可以监听指定DAG的最后一个任务完成状态作为前置判断条件，只需将ExternalTaskSensor的external_dag_id参数设置为目标子DAG的ID，external_task_id参数设置为对应子DAG最后一个任务的ID即可。
• 如果你不想绑定具体任务ID，也可以将external_task_id留空，此时传感器会直接监听整个外部DAG的运行状态，只要对应执行日期的外部DAG整体运行成功就会通过校验，更适配你需要确认前序DAG全量任务完成的需求。

ExternalTaskSensor导入失败的优先排查方案 #

大概率是Airflow版本不匹配导致的导入路径错误，你可以根据你使用的版本调整导入语句：

• Airflow 2.x版本正确导入路径：from airflow.sensors.external_task import ExternalTaskSensor
• Airflow 1.x版本正确导入路径：from airflow.contrib.sensors.external_task_sensor import ExternalTaskSensor
  调整路径后即可正常使用，这是实现需求成本最低的方案。

无法使用ExternalTaskSensor的替代方案 #

如果调整导入路径后仍无法使用该组件，可选择以下任意一种方案实现相同的调度逻辑：

方案1：自定义Python传感器查询DAG运行状态 #

直接查询Airflow元数据库的子DAG运行记录，判断是否完成，示例代码如下：

from airflow.sensors.python import PythonSensor
from airflow.models import DagRun

def check_sub_dag_finish(**context):
    target_dag_id = "你要监听的子DAG ID"
    # 匹配和主DAG同执行日期的子DAG运行实例
    dag_run_list = DagRun.find(dag_id=target_dag_id, execution_date=context["execution_date"])
    if dag_run_list and dag_run_list[0].state == "success":
        return True
    return False

# 传感器任务定义
wait_dag_a = PythonSensor(
    task_id="wait_dag_a_finish",
    python_callable=check_sub_dag_finish,
    poke_interval=60, # 每60秒轮询一次状态
    timeout=3600, # 超时时间可按需调整，单位为秒
    mode="reschedule", # 低功耗模式，避免长期占用worker槽位
    provide_context=True
)

方案2：基于输出产物的状态校验 #

如果你的Dataflow作业运行完成后会输出结果文件到对象存储（比如GCS、S3、OSS），可以直接使用对应存储的传感器（比如GCSObjectExistenceSensor、S3KeySensor）监听子DAG最终生成的结果文件/标记文件，文件存在即判定前序DAG运行完成。

方案3：用TaskGroup替代子DAG重构调度逻辑 #

如果所有子DAG的逻辑都在同一代码库维护，可以将原来的每个子DAG拆分为独立的TaskGroup，直接在同一个主DAG里定义串并行依赖，完全不需要跨DAG依赖，也规避了ExternalTaskSensor的使用问题，依赖管理更直观。

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