跨 DAG 依赖与就绪检查是什么
跨 DAG 依赖有两类思路:一是用 ExternalTaskSensor 等待上游某个 DAG 的指定任务在同一个逻辑日期成功;二是让下游直接检查上游的产出——分区是否就绪、标记文件是否存在。后者的优势是不依赖调度器的状态与上下游频率的对齐,上游换了调度方式、改了任务名,只要产出约定不变,下游依然跑得起来。
基本信息
| Sensor 方式 | ExternalTaskSensor 等待上游 DAG 与任务,需匹配逻辑日期 |
|---|---|
| 产出检查 | 查询目标分区行数或读取标记文件,不依赖调度器状态 |
| 时间对齐 | 上下游频率不同时需自定义日期映射函数(execution_date_fn) |
| 等待模式 | mode=reschedule 释放 worker 槽位,避免长时间占用 |
| 数据集方式 | Airflow 的数据集(3.x 为 Asset)也可作为跨 DAG 触发依据 |
| 超时设置 | 必须设 timeout 与适当的重试,否则会无限等待 |
典型使用场景
- 分层调度:ODS 完成后触发 DWD,DWD 完成后触发 DWS
- 跨团队依赖:下游检查上游发布的就绪标记,而非等待其任务状态
- 频率不一致:上游每小时、下游每天,用产出检查规避日期映射难题
- 外部系统就绪:等待第三方文件到达后再启动加载
- 关键路径保护:让下游在数据就绪后才启动,避免读空分区
ExternalTaskSensor 用法
ExternalTaskSensor 用法
from airflow.sensors.external_task import ExternalTaskSensor
wait_upstream = ExternalTaskSensor(
task_id="wait_upstream",
external_dag_id="ods_orders",
external_task_id="load",
allowed_states=["success"],
timeout=3600,
mode="reschedule", # 让出 worker 槽位,避免长期占用
)产出就绪检查(SQL 与标记文件)
产出就绪检查(SQL 与标记文件)
-- 分区就绪检查:行数为 0 说明上游还没产出,继续等待
SELECT COUNT(1) AS cnt
FROM ods_trade_order
WHERE dt = '2026-10-11';
-- 标记文件方式:上游完成后写出 _SUCCESS,下游判断其存在
-- 检查命令示例(在任务中执行):
-- hdfs dfs -test -e /warehouse/ods/trade_order/dt=2026-10-11/_SUCCESS两类方式对照
| 对比 | Sensor 方式 | 产出检查方式 |
|---|---|---|
| 依赖对象 | 上游任务状态 | 上游数据产出 |
| 频率不一致 | 需自定义日期映射 | 天然支持,按业务日期查即可 |
| 上游改名 | Sensor 等待失败 | 不受影响,只要产出约定不变 |
| 抢占资源 | 轮询会占用或反复检查 | 按间隔查询,开销可控 |
| 失效表现 | 超时失败,问题可见 | 一直等待,需配超时告警 |
上游任务改名会让 Sensor 永远等不到:发布上游改动前先搜一遍有哪些下游在依赖它,必要时把依赖改为产出检查以降低耦合。
只等上游「任务成功」而不校验数据是否落地,仍可能读到空分区:任务成功与产出可用之间可能有延迟,关键链路应增加行数或标记文件校验。