Apache Airflow Cron 时间表 DST 转换日修复：深入解析重复小时与折返（Fold）时段的调度语义

Apache Airflow Cron 时间表 DST 转换日修复：深入解析重复小时与折返（Fold）时段的调度语义

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

导读

本文围绕 Apache Airflow 3 中一个关键的时序 bug 修复展开：在夏令时（DST）回拨转换日，cron 时间表（timetable）可能为一个"墙钟上尚未经过"的时段提前创建 Dag run，导致数据间隔（data interval）错误。文章将以airflow-core/newsfragments/70089.bugfix.rst中记录的修复为线索，结合airflow-core/src/airflow/timetables/_cron.py与airflow-core/src/airflow/timetables/interval.py的源码实现，以及单元测试用例，完整还原问题的成因、修复原理与实际影响，帮助读者理解 Airflow 调度器在处理时区敏感调度时的内部机制，并掌握在 DST 场景下正确配置与验证 cron 调度的实战方法。

修复内容概述

本次修复的官方记录（airflow-core/newsfragments/70089.bugfix.rst）只有一句话：

Fix cron timetables that could schedule a Dag run for a period that has not elapsed yet on a DST transition day.

其含义是：在夏令时转换日，cron 时间表可能为一个"尚未经过"的时段调度 Dag run，此问题已被修复。这条 bugfix 片段属于 Airflow 的 newsfragments 机制——Apache Airflow 项目使用该目录（airflow-core/newsfragments）为每个 PR 记录变更说明，发布时由scripts/ci的 changelog 工具汇总到正式的 RELEASE_NOTES.rst 中。文件后缀.bugfix.rst表明这是一项 bug 修复类变更。

要理解这条修复的真正技术含量，需要先弄清楚两个问题：cron 时间表如何决定"何时可以创建 Dag run"，以及 DST 转换日为什么会打破这个判定。

背景：Cron 时间表与数据间隔（Data Interval）的调度逻辑

在 Airflow 2.2+（Airflow 3 沿用的架构）中，调度逻辑由Timetable（时间表）抽象负责。当你在 DAG 中写schedule="30 2 * * *"时，实际使用的是CronDataIntervalTimetable，其定义位于 airflow-core/src/airflow/timetables/interval.py：

class CronDataIntervalTimetable(CronMixin, _DataIntervalTimetable): """ Timetable that schedules data intervals with a cron expression. This corresponds to ``schedule=<cron>``, where ``<cron>`` is either a five/six-segment representation, or one of ``cron_presets``. The implementation extends on croniter to add timezone awareness. This is because croniter works only with naive timestamps, and cannot consider DST when determining the next/previous time. """

这段类文档透露了两个关键事实：

1. schedule=<cron>与CronDataIntervalTimetable一一对应，支持五/六段 cron 表达式以及cron_presets预置值（如@daily）。
2. croniter 只处理 naive（无时区）时间戳，无法感知 DST，因此 Airflow 必须在其基础上自行补齐时区感知逻辑——这正是本次修复所在的位置。

数据间隔的含义

Airflow 调度的核心产物是DataInterval（数据间隔）：一个 Dag run 通常处理"上一个调度点到当前调度点"之间的数据。CronDataIntervalTimetable.next_dagrun_info()根据上一次自动化运行的数据间隔终点（last_automated_data_interval.end）推导下一次运行的数据间隔：

• 若无上一次运行，则从TimeRestriction.earliest对齐后的起点出发；
• 有上一次运行时，先对齐上一次间隔终点（self._align_to_prev(last_automated_data_interval.end)），再以self._get_next(start)求出下一终点。

关键的不变量是：一个 Dag run 只应在它所对应的数据间隔"已经完全经过"之后才被创建。如果调度器把一个"墙钟上还没走完"的时段误判为已经结束，就会提前创建 Dag run，造成数据窗口错位、重复处理或空窗口。

问题本质：DST 转换日墙钟时间如何"欺骗"调度器

DST（夏令时）转换在每个使用夏令时的时区每年发生两次：

其中fold 是本次 bug 的重灾区。以源码注释中的例子（airflow-core/src/airflow/timetables/_cron.py）说明：瑞士 2023 年在 UTC+2 的凌晨 3 点退出 DST，把时钟回拨到 UTC+1 的凌晨 2 点。于是当地墙钟上会出现两个 02:00–03:00：

• 第一个 02:30 属于 UTC+2（DST 时段）；
• 第二个 02:30 属于 UTC+1（标准时段，即 fold 时段）。

如果 cron 表达式是30 * * * *（每小时的第 30 分钟），上一次运行在 02:30（UTC+2），下一次本应落在 fold 后的 02:30（UTC+1）——因为墙钟 02:30 在回拨后会再次出现。但 naive 的 croniter 无法区分这两个 02:30 谁先谁后，甚至可能直接跳到 03:30，导致：

• 调度器认为"02:30–03:30 这个墙钟时段"在第二个 02:30 尚未到达之前就已经"结束"；
• 于是在 UTC 时间戳上过早地为这个时段创建 Dag run——这就是 newsfragment 中所说的a period that has not elapsed yet（尚未经过的时段）。

另一个方向（进入 DST）同样有风险：如果调度时间落在被跳过的空洞小时（如0 2 * * *，而 02:00–03:00 被跳过），墙钟上根本不存在这个时刻，naive 计算可能把间隔终点映射到错误的 UTC 时刻，造成数据间隔整体偏移一小时。

源码级修复：CronMixin的时区感知与 fold 处理

本次修复的实质，是让CronMixin的_get_next/_get_prev在 croniter 返回 naive 时间后，以 UTC 时间戳为基准重新换算，并对"每小时至少运行一次"的高频 cron 应用 fold 特判。相关实现全部位于 airflow-core/src/airflow/timetables/_cron.py：

def _get_next(self, current: DateTime) -> DateTime: """Get the first schedule after specified time, with DST fixed.""" naive = make_naive(current, self._timezone) cron = croniter(self._expression, start_time=naive) scheduled = cron.get_next(datetime.datetime) if TYPE_CHECKING: assert isinstance(scheduled, datetime.datetime) if not _covers_every_hour(cron): return convert_to_utc(make_aware(scheduled, self._timezone)) delta = scheduled - naive return convert_to_utc(current.in_timezone(self._timezone) + delta) def _get_prev(self, current: DateTime) -> DateTime: """Get the first schedule strictly before specified time, with DST fixed.""" naive = make_naive(current, self._timezone) cron = croniter(self._expression, start_time=naive) scheduled = cron.get_prev(datetime.datetime) if TYPE_CHECKING: assert isinstance(scheduled, datetime.datetime) if not _covers_every_hour(cron): prev = convert_to_utc(make_aware(scheduled, self._timezone)) # croniter steps back on naive wall clock, but make_aware can map a tick inside # a DST transition forward onto current or later. Keep stepping until strictly # earlier; get_prev is strictly decreasing, so this terminates. while prev >= current: scheduled = cron.get_prev(datetime.datetime) prev = convert_to_utc(make_aware(scheduled, self._timezone)) return prev delta = naive - scheduled return convert_to_utc(current.in_timezone(self._timezone) - delta)

普通 cron：绝对时刻换算 + 严格单调回退

对于非每小时覆盖的 cron（例如0 2 * * *、*/15 * * * *之外的大多数表达式），修复逻辑是：

1. 把当前时刻current转成时间表时区的 naive 墙钟时间naive，交给 croniter 求下一个/上一个调度点；
2. 用make_aware(scheduled, self._timezone)把结果重新解释为该时区的带 fold 语义的时刻，再经convert_to_utc转回 UTC 绝对时刻。

这里make_aware对 fold 的处理是决定性的：当 croniter 返回的 naive 时间恰好落在重复小时（fold 区间）内时，make_aware会把fold=1固定在第二次出现的位置上，从而得到唯一的 UTC 映射。_get_prev中还额外加了一个防御循环：由于make_aware可能把折返区间内的某个 tick 映射到"当前或更晚"的 UTC 时刻（而不是更早），代码会持续调用cron.get_prev()回退，直到结果严格早于current为止——注释明确说明get_prev是严格递减的，因此该循环必然终止。这一改动直接保证了"上一个调度点"语义在任何 DST 边界上都不会越界。

每小时覆盖的 cron：fold 时区问题的特判

对于_covers_every_hour(cron)返回 True 的表达式（即 cron 的小时字段展开为*，如30 * * * *、*/30 * * * *），修复采用了时间差叠加策略（见 airflow-core/src/airflow/timetables/_cron.py 的_covers_every_hour与上文代码）：

def _covers_every_hour(cron: croniter) -> bool: """ Check whether the given cron runs at least once an hour. ... Folding happens when a region switches time backwards, usually as a part of ending a DST period, causing a block of time to occur twice in the wall clock. This is indicated by the ``fold`` flag on datetime. ... """ return cron.expanded[1] == ["*"]

其思路是：既然这种 cron 每小时都会触发，那么"上一次 02:30（UTC+2）→ 下一次 02:30（UTC+1）"之间的墙钟间隔恰好是一个小时（naive 运算得到delta = scheduled - naive），把这个 delta 叠加到current在该时区的绝对时刻上（current.in_timezone(self._timezone) + delta），再转回 UTC，就能精确命中 fold 后的第二次 02:30，而不是错误地跳到 03:30 或提前结束时段。_get_prev对称地使用naive - scheduled的 delta 做减法。

至于为什么只对每小时覆盖的 cron 做特判，源码注释给出了务实理由（airflow-core/src/airflow/timetables/_cron.py）：

While this technically happens for all cron schedules (in such a timezone), we only care about schedules that create at least one run every hour, and can provide a somewhat reasonable rationale to skip the fold hour for things such as*/2(every two hours). Therefore, we try tominimallypeek into croniter internals to work around the issue.

也就是说：fold 问题理论上影响该时区内所有 cron，但只有每小时至少运行一次的表达式才会在 fold 小时内部真正"踩中"重复时刻；对于*/2（每两小时）这类表达式，跳过 fold 小时是更合理的行为，因此用cron.expanded[1] == ["*"]这一最小侵入式判断来区分两条路径。

单元测试佐证：修复前后行为如何被验证

本次修复的正确性在 airflow-core/tests/unit/timetables/test_interval_timetable.py 中有成体系的测试保障，覆盖进入 DST、退出 DST、fold 区间、非平凡 cron 等全部场景。

测试一：test_fold_scheduling——每小时覆盖 cron 的 fold 全程

该测试（airflow-core/tests/unit/timetables/test_interval_timetable.py）使用CronDataIntervalTimetable("*/30 * * * *", timezone="Europe/Zurich")，从 2023-10-28 23:30 UTC（当地 DST 时间 01:30）开始逐步推进，完整走完"正常 → 跨入 fold → fold 区间内 → 走出 fold"四个阶段：

• 跨入 fold 时，间隔为[2023-10-29 00:30 UTC, 2023-10-29 01:00 UTC)，即当地 02:00（fold，而非 DST）；
• 在 fold 区间内，间隔为[2023-10-29 01:00 UTC, 2023-10-29 01:30 UTC)，仍是当地 02:00（fold）；
• 走出 fold 后，间隔为[2023-10-29 01:30 UTC, 2023-10-29 02:00 UTC)，当地 03:00（非 DST）。

注意这里数据间隔的终点都以 UTC 时刻唯一标识，fold 内的两个"02:00 当地时刻"被精确区分开来——这正是"不为尚未经过的时段创建 run"的直接验证：fold 后第二次 02:00 到达之前，调度器绝不会把该时段当作已完成。

测试二：TestCronIntervalDst——低频率 cron 的进出 DST 边界

该类（airflow-core/tests/unit/timetables/test_interval_timetable.py）针对0 2 * * */0 3 * * *这类每日 cron，验证Europe/Zurich时区 2023 年的两个转换：

• test_entering/test_entering_skip：进入 DST 时，由于当地 02:00 被跳过（不存在），间隔正确落在[01:00 UTC, 01:00 UTC)附近的边界上，而不是生成一个幻影的 02:00；
• test_exiting_exact/test_exiting_fold：退出 DST 时，间隔终点落在第二次 02:00（fold=1）对应的02:00 UTC上，与 Airflow 历史行为保持一致（测试注释明确说明："There are two 2am local times on the 29th due to folding. We end on the second one (fold=1). There's no logical reason here; this is simply what Airflow has been doing since a long time ago, and there's no point breaking it."）。

测试三：TestCronIntervalDstNonTrivial——区间型与多值 cron

该类（airflow-core/tests/unit/timetables/test_interval_timetable.py）复用了 2020 年洛杉矶的经典案例（源自 apache/airflow#7999），对0 7-8 * * *和0 7,9 * * *进行验证。核心断言是：跨越 DST 转换的那个数据间隔会"提前一小时结束"（测试注释："This interval ends an hour early since it includes the DST switch!"），随后回到正常间隔。这确认了修复对"每小时覆盖"之外的复杂表达式同样有效，且与历史行为兼容。

实战影响与验证方法

哪些 DAG 会受影响

• 受影响：使用 cron 表达式（含cron_presets）作为schedule，且 DAG 所在时区（default_timezone或 timetable 的timezone参数）实行夏令时的调度；
• 不受影响：使用timedelta/relativedelta的DeltaDataIntervalTimetable（见 interval.py）、OnceTimetable、DatasetTriggeredTimetable等非 cron 时间表，以及 UTC 时区（无 DST）的调度。

如何配置时区

cron 时间表的时区来源有两个层级（对应源码中CronMixin.__init__对timezone参数的解析，见 airflow-core/src/airflow/timetables/_cron.py）：

1. 全局默认：在airflow.cfg中设置default_timezone = Europe/Zurich（或环境变量AIRFLOW__CORE__DEFAULT_TIMEZONE）；
2. DAG 级覆盖：在 DAG 构造时传入schedule="30 2 * * *"并在 timetable 上显式指定时区。

需要提醒的是：调度器内部始终以 UTC 时间戳推进（源码中所有边界时刻最终都经convert_to_utc归一化），时区只影响"墙钟上几点触发"的解释，这正是 DST 问题必须在 timetable 层修复、而非在 croniter 层修复的根本原因。

如何验证

• 若本地可运行 Airflow 测试套件，直接运行 DST 相关用例：pytest airflow-core/tests/unit/timetables/test_interval_timetable.py -k "dst or fold"；
• 也可仿照test_fold_scheduling的写法，用CronDataIntervalTimetable构造目标时区与 cron，逐步调用next_dagrun_info断言数据间隔的 UTC 边界；
• 生产环境排查时，重点检查 DST 转换日（如每年 3 月最后一个周日与 10 月最后一个周日）附近 Dag run 的data_interval_start/data_interval_end是否为整点 UTC 对齐、是否存在时长不足一小时或重叠的间隔。

总结

70089.bugfix.rst修复的是 Airflow cron 时间表在 DST 转换日的时序错乱：由于 croniter 只能理解 naive 墙钟时间，在秋季回拨（fold）时调度器可能为一个尚未真正经过的重复时段提前创建 Dag run。修复在CronMixin层面补齐了时区感知——普通 cron 通过make_aware的 fold 语义加严格单调回退保证边界唯一性，每小时覆盖的 cron 通过墙钟 delta 叠加精确落在 fold 后的第二次触发点，并辅以_covers_every_hour的最小侵入式分流。相关行为已由 test_interval_timetable.py 中覆盖进出 DST、fold 全程与非平凡表达式的多组测试锁定。对于任何运行在夏令时时区的 Airflow 用户，理解这一修复，能帮助你在 DST 转换日精准判断数据窗口是否正确，避免因重复小时或空洞小时引发的数据重复、漏处理与空窗口问题。

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