数据集驱动调度是什么
数据集（Airflow 2.4 引入的 Dataset，3.x 改名 Asset）把「上游产出了什么」作为下游 DAG 的触发条件，替代按时间硬对齐的 cron 表达式。生产任务用 outlets 声明自己产出某个数据集，下游 DAG 把该数据集写进 schedule，一旦有任务更新这个数据集就会触发一次下游运行。同一数据集被多个 DAG 更新时，下游会等所有更新到齐才触发，这正好解决了「上游不定时完成」的对齐难题。
基本信息
| 版本差异 | Airflow 2.4 起为 Dataset，3.x 中更名为 Asset |
|---|---|
| 声明方式 | 下游 DAG 的 schedule 传数据集列表 |
| 生产者标记 | 任务参数 outlets 声明该任务产出哪些数据集 |
| 触发语义 | 数据集被更新即产生一次下游运行；多个上游等齐再触发 |
| 与 Sensor 差异 | 不依赖上游任务状态与逻辑日期对齐，只看产出是否更新 |
| 兜底要求 | 仍需配超时告警，防止上游整天无产出导致下游静默不跑 |
典型使用场景
- 上游不定时完成：源端文件到点不定，产出即触发下游，无需按点轮询
- 多上游汇聚：多个上游 DAG 都产出同一数据集，等齐后再跑下游汇总
- 跨团队交接：以数据集为接口约定，减少对彼此调度时间的依赖
- 替代高频轮询：不再频繁检查上游状态，降低调度器与元数据压力
- 产出驱动的推送链路：上游完成即推，缩短端到端延迟
数据集声明示例
数据集声明示例
# 生产者：任务用 outlets 声明产出的数据集
@task(outlets=[orders_ds])
def produce_orders():
pass
# 消费者：DAG 以数据集为触发条件
@dag(schedule=[orders_ds], start_date=datetime(2026, 1, 1))
def downstream():
pass核对触发链路
核对触发链路
# 查看数据集定义与关联的 DAG（命令名随版本略有差异）
airflow datasets list
# 查看由数据集触发的运行记录，确认是否按期产生
airflow dags list-runs -d downstream | head两种触发方式对照
| 对比 | 数据集驱动 | 定时与 Sensor |
|---|---|---|
| 触发依据 | 上游是否更新了数据集 | 时间点或上游任务状态 |
| 上游不定时 | 天然适配，产出即触发 | 需要轮询或改调度时间 |
| 逻辑日期对齐 | 不依赖日期映射 | 频率不同需自定义映射函数 |
| 轮询开销 | 无轮询，事件驱动 | 轮询会持续占用或反复检查 |
| 失效表现 | 下游完全不触发，无报错 | Sensor 超时后才失败 |
数据集标识写错（路径或命名空间与生产者不一致）会让下游永远不触发，而且不会报错：排查这类「什么都不跑」的问题时，先核对生产者的 outlets 与下游 schedule 是否指向同一个标识。
数据集触发只说明产出动作发生了，不校验内容是否正确：数据是否为空、是否完整仍要靠行数比对与质量规则来兜底。