Apache Airflow 自定义 Timetable 调度指南：以“工作日下班后“定时任务为例-创锋一号

Apache Airflow 自定义 Timetable 调度指南：以"工作日下班后"定时任务为例

【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow

本文基于 Apache Airflow 官方 HowTo 文档 airflow-core/docs/howto/timetable.rst 编写，核心示例代码来自仓库中的 workday.py，并参考了 airflow-core/src/airflow/timetables/base.py、airflow-core/src/airflow/timetables/interval.py 等源码，全文基于 Airflow 3.x 仓库实际内容撰写。

时间信息（2026-09-09）为当前环境时间，文中所有涉及日期、版本的陈述均以本仓库代码与文档为唯一依据。

从需求说起：为什么需要 Timetable
Timetable 核心接口与基础类型
Timetable 注册与 DAG 接入
实现调度逻辑：两个核心方法
参数化 Timetable 与序列化
Timetable 在 UI 中的展示：summary 与 description
自定义 generate_run_id
测试验证：以 test_workday_timetable.py 为例
延伸阅读

从需求说起：为什么需要 Timetable

Airflow 内置的 cron 表达式调度能力非常强大，但无法覆盖所有业务场景。本文以一个典型需求为例：一家公司希望在每个工作日结束后，处理当天工作时间内采集的数据。

直觉上，大家首先会想到 cron 表达式schedule="0 0 * * 1-5"（周一至周五的午夜零点）。但这个方案有两个明显缺陷：

数据积压问题：周五采集的数据不会在周五结束时立刻处理，而是要等到下周一零点，且这次运行的 interval 会跨越"周五零点到下周一零点"整整三天；
无法跳过节假日：cron 表达式无法感知法定节假日，节假日当天依然会触发调度。

因此我们真正想要的是：

在每个周一、周二、周三、周四、周五各调度一次运行，每次运行的 data interval 覆盖当天零点到次日零点（例如 2021-01-01 00:00:00 至 2021-01-02 00:00:00）；
每次运行在 data interval 结束之后立刻创建：覆盖周一的运行在周二零点触发，覆盖周五的运行在周六零点触发；周日和周一零点不产生运行；
在定义的节假日不调度运行。

为简化说明，本文示例只使用 UTC 时区的 datetime。

Timetable 核心接口与基础类型

在动手实现前，需要先了解 airflow-core/src/airflow/timetables/base.py 中定义的几个核心类型。它们是所有自定义 Timetable 的"语言"：

DataInterval：一个NamedTuple，包含start和end两个字段，描述一次 DagRun 所处理的数据区间。其exact(at)类方法可以构造一个只包含单个时刻的"零宽"区间；
TimeRestriction：一个NamedTuple，封装了 DAG 及其任务对调度时间的所有限制，包含三个字段：
- earliest：DAG 可以被调度的最早时间，由 DAG 和所有任务的start_date参数计算得出，如果没有start_date则为None；
- latest：与earliest类似，由end_date参数计算得出；
- catchup：布尔值，反映 DAG 的catchup参数，默认为False；
DagRunInfo：一个NamedTuple，描述一次 DagRun 的调度信息，包含run_after（DagRun 最早可以被创建并调度任务的时间）和data_interval两个字段。其interval(start, end)类方法用于构造"区间结束即触发"的运行，且data_interval.end == run_after恒成立；
Timetable：一个Protocol，是所有 Timetable 类需要实现的接口，其中最重要的两个方法是next_dagrun_info和infer_manual_data_interval。

# DagRunInfo 的两种构造方式 info = DagRunInfo( data_interval=DataInterval(start=start, end=end), run_after=run_after, ) # 通常我们希望在区间结束时立即触发，因此有更简洁的快捷方式 info = DagRunInfo.interval(start=start, end=end) assert info.data_interval.end == info.run_after # Always True.

关于DataInterval中的 datetime，有一条硬性要求：所有由自定义 Timetable 返回的 datetime 值必须是 "aware" 的（即包含时区信息），且必须使用pendulum的 datetime 和 timezone 类型。这一点在 base.py 的DataInterval定义和文档的 note 中都有明确强调。

此外，Timetable还定义了一些属性，自定义实现可以根据需要覆盖：

description: str = ""：人类可读的 Timetable 描述，用于 Web UI 展示（如 cron 表达式'30 21 * * 5'可描述为'At 21:30, only on Friday'）；
periodic: bool = True：该 Timetable 是否周期性运行（schedule=None和"@once"等特殊设置会将其置为False）；
can_be_scheduled: bool = True：能否真正以自动化方式调度运行（NullTimetable会将其置为False）；
run_ordering：该 Timetable 触发的运行在 UI 中的排序字段，默认为("data_interval_end", "logical_date")；
active_runs_limit: int | None = None：DAG 同时可处于活跃状态的最大运行数（在 DAG 初始化时调用，返回值用作 DAG 的默认max_active_runs）；
summary属性：用于在 Web UI 中展示 Timetable 的简短摘要，默认实现返回类的类型名；
type_name属性：主要用于按 Timetable 类型过滤 DAG，内置 Timetable 返回类名，自定义 Timetable 返回完整导入路径；
validate()方法：在 DAG 放入 dagbag 时进行运行时校验，失败时抛出AirflowTimetableInvalid；
serialize()/deserialize()：DAG 序列化与反序列化时使用（详见下文"参数化 Timetable"一节）。

Timetable 注册与 DAG 接入

自定义 Timetable 必须继承airflow.timetables.base.Timetable，并作为 plugin 的一部分注册。下面是实现新 Timetable 的骨架：

from airflow.plugins_manager import AirflowPlugin from airflow.timetables.base import Timetable class AfterWorkdayTimetable(Timetable): pass class WorkdayTimetablePlugin(AirflowPlugin): name = "workday_timetable_plugin" timetables = [AfterWorkdayTimetable]

实现完成后，就可以在 DAG 文件中使用这个 Timetable：

import pendulum from airflow.sdk import DAG from airflow.example_dags.plugins.workday import AfterWorkdayTimetable with DAG( dag_id="example_after_workday_timetable_dag", start_date=pendulum.datetime(2021, 3, 10, tz="UTC"), schedule=AfterWorkdayTimetable(), tags=["example", "timetable"], ): ...

仓库中 workday.py 的WorkdayTimetablePlugin正是通过timetables = [AfterWorkdayTimetable]这种方式把自定义 Timetable 注册进 Airflow 的插件系统。

实现调度逻辑：两个核心方法

当 Airflow 的 scheduler 遇到一个 DAG 时，会调用以下两个方法之一来决定何时调度该 DAG 的下一次运行：

next_dagrun_info：scheduler 用它来了解 Timetable 的常规调度节奏，即本例中"每个工作日一次、在工作日结束时运行"的部分；
infer_manual_data_interval：当 DagRun 被手动触发（例如从 Web UI 触发）时，scheduler 用该方法反向推断这个"计划外"运行的数据区间。

我们首先实现较简单的infer_manual_data_interval。仓库 workday.py 中的完整实现如下：

# [START howto_timetable_infer_manual_data_interval] def infer_manual_data_interval(self, run_after: DateTime) -> DataInterval: start = DateTime.combine((run_after - timedelta(days=1)).date(), Time.min).replace(tzinfo=UTC) # Skip backwards over weekends and holidays to find last run start = self.get_next_workday(start, incr=-1) return DataInterval(start=start, end=(start + timedelta(days=1))) # [END howto_timetable_infer_manual_data_interval]

该方法接受一个参数run_after（一个pendulum.DateTime对象），表示 DAG 被外部触发的时间。由于我们的 Timetable 为每个完整的"工作日"创建一个数据区间，这里推断出的数据区间通常应从run_after的前一天午夜开始；但如果run_after落在周日或周一（即前一天是周六或周日），则应该继续向前推到上一个周五。一旦确定了区间的起点，终点就是起点之后完整的一天。最后创建一个DataInterval对象来描述这个区间。

接下来是next_dagrun_info的实现。仓库 workday.py 中的完整实现如下：

# [START howto_timetable_next_dagrun_info] def next_dagrun_info( self, *, last_automated_data_interval: DataInterval | None, restriction: TimeRestriction, ) -> DagRunInfo | None: if last_automated_data_interval is not None: # There was a previous run on the regular schedule. last_start = last_automated_data_interval.start next_start = DateTime.combine((last_start + timedelta(days=1)).date(), Time.min) # Otherwise this is the first ever run on the regular schedule... elif (earliest := restriction.earliest) is None: return None # No start_date. Don't schedule. elif not restriction.catchup: # If the DAG has catchup=False, today is the earliest to consider. next_start = max(earliest, DateTime.combine(Date.today(), Time.min, tzinfo=UTC)) elif earliest.time() != Time.min: # If earliest does not fall on midnight, skip to the next day. next_start = DateTime.combine(earliest.date() + timedelta(days=1), Time.min) else: next_start = earliest # Skip weekends and holidays next_start = self.get_next_workday(next_start.replace(tzinfo=UTC)) if restriction.latest is not None and next_start > restriction.latest: return None # Over the DAG's scheduled end; don't schedule. return DagRunInfo.interval(start=next_start, end=(next_start + timedelta(days=1))) # [END howto_timetable_next_dagrun_info]

该方法接受两个参数：

last_automated_data_interval：一个DataInterval实例，表示该 DAG 上一次非手动触发运行的数据区间；如果这是该 DAG 有史以来第一次被调度，则为None。注意：last_automated_data_interval只在 DAG 第一次被 Dag processor 拾取时为None——首次运行在解析时就被计算出来并存储在 DAG 上。在调度阶段，next_dagrun_info总是带着上一次运行的数据区间被调用，因此 DAG 首次取消暂停时，scheduler 日志中不会出现None的情况（base.py 的 docstring 也有同样说明）；
restriction：封装了 DAG 及其任务对调度规格的限制，即上面提到的TimeRestriction。

一个容易被忽略的细节是：earliest和latest作用于 DagRun 的 logical date（即数据区间的起点），而不是运行被调度的时间（通常晚于数据区间结束）。

调度逻辑分情况讨论：

如果之前已经有过一次按常规调度运行的记录：基于上一次运行的start加一天，作为下一个候选起点，然后跳过周末和节假日；
如果这是首次调度，且restriction.earliest为None：说明 DAG 没有设置start_date，直接返回None不调度；
如果catchup为False：即便start_date在过去，也不能调度当前时间之前的运行，取earliest与"今天零点"中的较晚者作为候选起点；
如果earliest不在午夜：跳到下一天的零点；
其他情况：直接以earliest作为候选起点。

之后通过get_next_workday跳过周末和节假日。最后，如果计算出的数据区间起点晚于restriction.latest，必须遵守限制，返回None表示不调度。

关键辅助方法get_next_workday的实现（workday.py）：

def get_next_workday(self, d: DateTime, incr=1) -> DateTime: holiday_calendar = self._get_holiday_calendar() next_start = d while True: if next_start.weekday() not in (5, 6): # not on weekend if holiday_calendar is None: holidays = set() else: holidays = holiday_calendar.holidays(start=next_start, end=next_start).to_pydatetime() if next_start not in holidays: break next_start = next_start.add(days=incr) return next_start

它通过循环递增/递减天数，跳过周六（weekday=5）、周日（weekday=6）以及节假日，找到下一个（或上一个，当incr=-1时）工作日。节假日日历使用pandas.tseries.holiday.USFederalHolidayCalendar（美国联邦节假日），采用惰性加载并缓存到类属性_holiday_calendar中；如果pandas导入失败，则打印 warning 并退化为不处理节假日。

为方便读者对照，这里给出 plugin 和 DAG 文件的完整参考（workday.py）：

# [START howto_timetable] from pendulum import UTC, Date, DateTime, Time from airflow.plugins_manager import AirflowPlugin from airflow.timetables.base import DagRunInfo, DataInterval, Timetable if TYPE_CHECKING: from airflow.timetables.base import TimeRestriction class AfterWorkdayTimetable(Timetable): _NOT_LOADED = object() _holiday_calendar = _NOT_LOADED @classmethod def _get_holiday_calendar(cls): if cls._holiday_calendar is cls._NOT_LOADED: try: from pandas.tseries.holiday import USFederalHolidayCalendar cls._holiday_calendar = USFederalHolidayCalendar() except ImportError: log.warning("Could not import pandas. Holidays will not be considered.") cls._holiday_calendar = None return cls._holiday_calendar def get_next_workday(self, d: DateTime, incr=1) -> DateTime: # ...（见上文） def infer_manual_data_interval(self, run_after: DateTime) -> DataInterval: # ...（见上文） def next_dagrun_info( self, *, last_automated_data_interval: DataInterval | None, restriction: TimeRestriction, ) -> DagRunInfo | None: # ...（见上文） class WorkdayTimetablePlugin(AirflowPlugin): name = "workday_timetable_plugin" timetables = [AfterWorkdayTimetable] # [END howto_timetable]

对应的 DAG 文件：

import pendulum from airflow.sdk import DAG from airflow.example_dags.plugins.workday import AfterWorkdayTimetable from airflow.providers.standard.operators.empty import EmptyOperator with DAG( dag_id="example_workday_timetable", start_date=pendulum.datetime(2021, 1, 1, tz="UTC"), schedule=AfterWorkdayTimetable(), tags=["example", "timetable"], ): EmptyOperator(task_id="run_this")

参数化 Timetable 与序列化

有时我们需要向 Timetable 传递一些运行时参数。继续以AfterWorkdayTimetable为例：假设有些 DAG 运行在不同的时区，我们希望某些 DAG 在第二天早上 8 点而不是午夜触发。与其为每种用途单独创建一个 Timetable，不如让 Timetable 接受参数：

class SometimeAfterWorkdayTimetable(Timetable): def __init__(self, schedule_at: Time) -> None: self._schedule_at = schedule_at def next_dagrun_info(self, last_automated_dagrun, restriction): ... end = start + timedelta(days=1) return DagRunInfo( data_interval=DataInterval(start=start, end=end), run_after=DateTime.combine(end.date(), self._schedule_at).replace(tzinfo=UTC), )

如果要把AfterWorkdayTimetable的首次运行逻辑适配为自定义的schedule_at值，需要注意将候选时间与self._schedule_at比较。前面示例中"仅在午夜调度"的检查，只在运行于00:00触发时才是正确的。例如：earliest为06:00时，应该仍然允许当天08:00的运行；而earliest为09:00时，则应顺延到下一个工作日。

由于 Timetable 是 DAG 的一部分，需要告诉 Airflow 如何结合__init__中提供的上下文对它进行序列化。这通过在 Timetable 类上实现两个额外方法来完成：

class SometimeAfterWorkdayTimetable(Timetable): ... def serialize(self) -> dict[str, Any]: return {"schedule_at": self._schedule_at.isoformat()} @classmethod def deserialize(cls, value: dict[str, Any]) -> Timetable: return cls(Time.fromisoformat(value["schedule_at"]))

DAG 被序列化时，会调用serialize获得一个可 JSON 序列化的值；当 scheduler 访问序列化后的 DAG 时，该值被传递给deserialize，用于重建 Timetable。base.py 中默认的deserialize无参构造类、默认的serialize返回空字典。内置实现中，CronDataIntervalTimetable 将 cron 表达式与时区序列化为{"expression": ..., "timezone": ...}，DeltaDataIntervalTimetable 则序列化{"delta": ...}，可作为参照。

Timetable 在 UI 中的展示：summary 与 description

默认情况下，自定义 Timetable 在 UI 中（例如 "dags" 表格的Schedule列）显示其类名。可以通过重写summary属性来自定义展示，这对于参数化 Timetable 特别有用，可以把__init__中传入的参数展示出来。对于SometimeAfterWorkdayTimetable类，可以这样写：

@property def summary(self) -> str: return f"after each workday, at {self._schedule_at}"

于是对于如下声明的 DAG：

with DAG( schedule=SometimeAfterWorkdayTimetable(Time(8)), # 8am. ..., ): ...

Schedule列会显示after each workday, at 08:00:00。

summary的默认实现（base.py）返回类型的类名；内置的CronMixin则返回 cron 表达式本身（airflow-core/src/airflow/timetables/_cron.py）。

此外，还可以通过重写description属性为 Timetable 提供更完整的描述。这在 UI 中展示全面描述时特别有用。对于SometimeAfterWorkdayTimetable类，可以这样写：

description = "Schedule: after each workday"

如果希望根据构造参数动态派生描述，也可以把description放到__init__里：

def __init__(self) -> None: self.description = "Schedule: after each workday, at f{self._schedule_at}"

这在需要提供与summary属性不同的全面描述时特别有用。以上述 DAG 为例，UI 中i图标会显示Schedule: after each workday, at 08:00:00。

内置实现中，CronMixin 的__init__会用ExpressionDescriptor（cron-descriptor 库）将 cron 表达式翻译成自然语言描述，例如'30 21 * * 5'会被描述为'At 21:30, only on Friday'；当 DOM 与 DOW 同时受限时，还会把冲突场景描述为 "OR" 语义（如'day-of-month desc (or) day-of-week desc'），并在解析失败时将description置为空字符串。CronDataIntervalTimetable的 description 实现可参考 airflow-core/src/airflow/timetables/interval.py。

自定义 generate_run_id

自 Airflow 2.4 起，Timetable 也负责为 DagRun 生成run_id。

例如，如果希望 Run ID 显示"人类友好"的运行开始日期（即数据区间结束的日期，而不是目前使用的区间起始日期），可以在自定义 Timetable 中添加如下方法：

def generate_run_id( self, *, run_type: DagRunType, logical_date: DateTime, data_interval: DataInterval | None, **extra, ) -> str: if run_type == DagRunType.SCHEDULED and data_interval: return data_interval.end.format("YYYY-MM-DD dddd") return super().generate_run_id( run_type=run_type, logical_date=logical_date, data_interval=data_interval, **extra )

注意：

run_id长度限制为250 个字符；
同一个 DAG 内的run_id必须唯一。

Timetable协议中generate_run_id的默认实现（base.py）是run_type.generate_run_id(suffix=run_after.isoformat())。除generate_run_id外，接口还提供next_dagrun_info_v2（next_dagrun_info的包装，从DagRunInfo提取 data interval）以及run_info_from_dag_run/next_run_info_from_dag_model等辅助方法，供 scheduler 在不同场景使用，感兴趣可以继续阅读 base.py。

测试验证：以 test_workday_timetable.py 为例

仓库提供了配套的单元测试 airflow-core/tests/unit/timetables/test_workday_timetable.py，可以帮助理解上述调度逻辑的正确行为：

test_first_schedule：由于 DAG 的start_date是 2021-09-04（周六），且第一个周一（2021-09-06）是美国节假日（Labor Day），所以第一次运行覆盖的是下周二（2021-09-07），并在周三触发，即DagRunInfo.interval(2021-09-07, 2021-09-08)；
test_subsequent_weekday_schedule：参数化测试验证接下来四次的运行分别覆盖后续四个工作日，每个 interval 为[day, day+1天)；
test_next_schedule_after_friday：周五的运行之后，下一次运行覆盖的是下周一，验证了跨周末跳过的行为；
test_holiday_calendar_is_cached：验证节假日日历只初始化一次并被复用；
test_holiday_calendar_falls_back_to_none_on_import_error：验证 pandas 导入失败时日历退化为None（即不处理节假日）。

这些测试直接从airflow.example_dags.plugins.workday导入AfterWorkdayTimetable，并与airflow.timetables.base中的DagRunInfo、DataInterval、TimeRestriction交互，构成了一个完整的"文档 → 示例实现 → 测试验证"闭环。

企业官网建设流程全解析