导读（Introduction）
欢迎来到 Apache Airflow 源码深度解析系列的第八课。
在前面的课程中，我们了解了 Scheduler 如何管理 DAG 的生命周期、如何将任务分发给 Executor 执行。但有一个核心问题始终在背后支撑着整个调度系统的运转——Scheduler 如何决定一个 DAG 应该在什么时候运行？答案就是本课的主角：Timetable（时间表）机制。
Timetable 是 Airflow 调度系统的"时间引擎"。从早期的schedule_interval字符串，到 Airflow 2.2 引入的 Timetable 抽象接口，再到 Airflow 3.x 中进一步丰富的触发式调度和分区调度，这一机制经历了深刻的架构演进。理解 Timetable 不仅能帮助你正确配置 DAG 的调度策略，更能让你在需要复杂调度逻辑（如跳过节假日、多时区调度、基于数据资产的事件驱动）时，自信地实现自定义 Timetable。
本课将从 Timetable Protocol 的设计哲学出发，深入分析DataInterval和DagRunInfo的核心概念，逐一剖析内置 Timetable 的实现，最后通过一个完整的"工作日调度"示例展示如何扩展 Airflow 的时间系统。
学习目标（Learning Objectives）
完成本课学习后，你将能够：
- 理解从 schedule_interval 到 Timetable 的演进历程——明确为什么需要 Timetable 抽象
- 掌握 DataInterval 和 DagRunInfo 的核心概念——理解数据区间与调度时间的关系
- 深入分析 Timetable Protocol 的接口设计——掌握
next_dagrun_info、infer_manual_data_interval等关键方法 - 理解 CronMixin 和 DeltaMixin 的实现细节——明确 Cron 表达式和时间增量的调度算法
- 掌握 DataInterval 模式与 Trigger 模式的本质区别——两种调度范式的适用场景
- 了解 Asset-Triggered 和 Partition 调度——事件驱动与分区式调度的新范式
- 具备实现自定义 Timetable 的能力——通过工作日调度实例掌握扩展方法
正文内容（Main Content）
1. 从 schedule_interval 到 Timetable 的演进
1.1 早期的 schedule_interval
在 Airflow 2.2 之前，DAG 的调度时间通过schedule_interval参数定义：
# Airflow 2.x 早期方式dag=DAG(dag_id="my_dag",schedule_interval="0 2 * * *",# 每天凌晨 2 点start_date=datetime(2024,1,1),)这种方式虽然简单直观，但存在几个根本性限制：
| 限制 | 说明 |
|---|---|
| 表达力不足 | 无法表达"每个工作日"、"每月最后一个交易日"等复杂规则 |
| 数据区间不透明 | execution_date的含义令人困惑——它实际是数据区间的起始而非执行时间 |
| 无法解耦调度与数据 | 调度周期和数据区间强制绑定为相同间隔 |
| 不可扩展 | 用户只能使用预定义的 Cron 表达式或 timedelta |
1.2 AIP-39：Timetable 的诞生
AIP-39（Richer scheduler_interval）是 Airflow 架构演进中的一个里程碑提案。它引入了两个核心概念：
- Timetable：一个可插拔的抽象接口，负责计算"下一次 DAG 运行应该在什么时候"
- DataInterval：显式的数据区间概念，将"这次运行处理哪段时间的数据"与"这次运行在什么时候执行"明确分离
这意味着：
execution_date被logical_date（DataInterval.start）取代，语义更清晰- 用户可以实现任意复杂的调度逻辑，只需遵循 Timetable 协议
- 调度周期和数据区间可以独立配置
1.3 Airflow 3.x 的进一步演进
在 Airflow 3.x 中，Timetable 机制进一步增强：
- Trigger 模式 Timetable：区分"数据区间模式"和"触发模式"，后者在调度时间点立即运行
- Asset-Triggered Timetable：基于数据资产事件驱动调度
- Partition Timetable：基于分区键的调度（AIP-76），支持分区级别的增量处理
- SDK 层与 Core 层分离：SDK 定义轻量级 Timetable 类型标识，Core 层实现完整调度逻辑
- Multiple Cron 表达式：支持在多个 Cron 时间点触发同一个 DAG
2. 核心概念：DataInterval 与 DagRunInfo
2.1 DataInterval：数据区间
DataInterval是理解 Airflow 调度语义的关键。它明确回答了一个问题：这次 DAG 运行处理的是哪个时间段的数据？
# 源码位置: airflow-core/src/airflow/timetables/base.py:42classDataInterval(NamedTuple):""" A data interval for a DagRun to operate over. Both ``start`` and ``end`` **MUST** be "aware", i.e. contain timezone information. """start:DateTime end:DateTime@classmethoddefexact(cls,at:DateTime)->DataInterval:"""Represent an "interval" containing only an exact time."""returncls(start=at,end=at)关键设计点：
start和end都必须是时区感知的（timezone-aware）日期时间对象- 区间是左闭右开语义：
[start, end)，即包含 start 但不包含 end exact()工厂方法用于表示"点时间"事件（如手动触发、@once 调度）
示例理解：
对于 schedule="0 0 * * *"（每天午夜）的 DAG： 第一次运行: DataInterval(start=2024-01-01 00:00, end=2024-01-02 00:00) → 意味着这次运行处理 1月1日 整天的数据 第二次运行: DataInterval(start=2024-01-02 00:00, end=2024-01-03 00:00) → 意味着这次运行处理 1月2日 整天的数据 logical_date = DataInterval.start = 2024-01-01 00:00 run_after = DataInterval.end = 2024-01-02 00:00 # 实际执行时间这解释了为什么"1月1日的数据"要等到"1月2日凌晨"才处理——因为只有当一天结束后，这一天的数据才是完整的。
2.2 DagRunInfo：调度信息
DagRunInfo是 Timetable 返回给 Scheduler 的调度决策结果：
# 源码位置: airflow-core/src/airflow/timetables/base.py:110classDagRunInfo(NamedTuple):"""Information to schedule a DagRun."""run_after:DateTime"""The earliest time this DagRun is created and its tasks scheduled."""data_interval:DataInterval|None"""The data interval this DagRun to operate over."""partition_date:DateTime|Nonepartition_key:str|None@classmethoddefexact(cls,at:DateTime)->DagRunInfo:"""Represent a run on an exact time."""returncls(run_after=at,data_interval=DataInterval.exact(at),partition_key=None,partition_date=None,)@classmethoddefinterval(cls,start:DateTime,end:DateTime)->DagRunInfo:"""Represent a run on a continuous schedule."""returncls(run_after=end,data_interval=DataInterval(start,end),partition_key=None,partition_date=None,)各字段含义：
| 字段 | 含义 |
|---|---|
run_after | 这个 DagRun 最早可以被创建的时间（Scheduler 等到这个时间后才创建 DagRun） |
data_interval | 这次运行覆盖的数据区间 |
partition_date | 分区日期（用于分区式调度） |
partition_key | 分区键（用于分区式调度） |
两个工厂方法的语义区别：
DagRunInfo.interval(start, end)：数据区间模式，run_after=end——即等到区间结束后才运行DagRunInfo.exact(at)：触发模式，run_after=at——在指定时间点立即运行
2.3 TimeRestriction：时间约束
Scheduler 在询问 Timetable “下一次什么时候运行” 时，会传入一个TimeRestriction约束：
# 源码位置: airflow-core/src/airflow/timetables/base.py:59classTimeRestriction(NamedTuple):"""Restriction on when a DAG can be scheduled for a run."""earliest:DateTime|None# 最早允许调度的时间（通常是 start_date）latest:DateTime|None# 最晚允许调度的时间（通常是 end_date）catchup:bool# 是否需要回填错过的运行这三个字段来自 DAG 定义中的start_date、end_date和catchup参数，共同限制了 Timetable 可以产生的调度范围。
3. Timetable Protocol 接口设计
3.1 Protocol 模式而非 ABC
Airflow 3.x 中的Timetable使用 Python 的Protocol（协议）而非传统的抽象基类（ABC）。这是一个有意的设计选择：
# 源码位置: airflow-core/src/airflow/timetables/base.py:167@runtime_checkableclassTimetable(Protocol):"""Protocol that all Timetable classes are expected to implement."""description:str=""periodic:bool=Truecan_be_scheduled:bool=Truerun_ordering:Sequence[str]=("data_interval_end","logical_date")active_runs_limit:int|None=Noneasset_condition:SerializedAssetBase=_NullAsset()partitioned:bool=False使用 Protocol 的优势：
- 结构化子类型：无需显式继承，只要实现了协议方法的类都被视为 Timetable
- SDK/Core 分离：SDK 层可以定义轻量级 Timetable 而无需依赖 Core 的完整实现
- 向后兼容：第三方库可以独立实现 Timetable 而不引入 Airflow 依赖
3.2 关键属性解析
| 属性 | 默认值 | 含义 |
|---|---|---|
description | "" | 人类可读的调度描述（如 “At 21:30, only on Friday”） |
periodic | True | 是否周期性运行。@once和None设为 False |
can_be_scheduled | True | 是否能自动调度。NullTimetable设为 False |
run_ordering | ("data_interval_end", "logical_date") | UI 中运行的排序字段 |
active_runs_limit | None | 最大活跃运行数限制。@continuous设为 1 |
asset_condition | _NullAsset() | 触发此 DAG 的数据资产条件 |
partitioned | False | 是否为分区式调度 |
3.3 核心方法
next_dagrun_info—— 最核心的方法：
defnext_dagrun_info(self,*,last_automated_data_interval:DataInterval|None,restriction:TimeRestriction,)->DagRunInfo|None:""" Provide information to schedule the next DagRun. :param last_automated_data_interval: The data interval of the associated DAG's last scheduled or backfilled run (manual runs not considered). :param restriction: Restriction to apply when scheduling the DAG run. :return: Information on when the next DagRun can be scheduled. None means a DagRun will not happen. """Scheduler 在每次心跳循环中调用此方法。传入上一次自动运行的数据区间和时间约束，Timetable 返回下一次运行的调度信息，或返回None表示不再需要新的运行。
infer_manual_data_interval—— 处理手动触发：
definfer_manual_data_interval(self,*,run_after:DateTime)->DataInterval:"""When a DAG run is manually triggered, infer a data interval for it."""当用户通过 UI 或 API 手动触发 DAG 时，Timetable 需要根据触发时间推断一个合理的数据区间。不同的 Timetable 策略各异——Cron 类型会取最近完成的周期，Delta 类型会取[run_after - delta, run_after]。
serialize/deserialize—— 序列化支持：
defserialize(self)->dict[str,Any]:"""Serialize the timetable for JSON encoding."""return{}@classmethoddefdeserialize(cls,data:dict[str,Any])->Timetable:"""Deserialize a timetable from data."""returncls()Timetable 的配置需要随 DAG 一起序列化存储到数据库。Cron 表达式、时区、间隔等参数通过这对方法持久化和恢复。
4. 内置 Timetable 类型详解
Airflow 3.x 提供了丰富的内置 Timetable 实现，可以分为四大类别：
4.1 简单调度类（simple.py）
这些 Timetable 处理最基本的调度场景，共享_TrivialTimetable基类：
# 源码位置: airflow-core/src/airflow/timetables/simple.py:59class_TrivialTimetable(Timetable):"""Some code reuse for "trivial" timetables that has nothing complex."""periodic=Falserun_ordering:Sequence[str]=("logical_date",)NullTimetable（schedule=None）：
# 源码位置: airflow-core/src/airflow/timetables/simple.py:88classNullTimetable(_TrivialTimetable):"""Timetable that never schedules anything."""can_be_scheduled=Falsedescription:str="Never, external triggers only"defnext_dagrun_info
