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.2Airflow 3 沿用的架构中调度逻辑由Timetable时间表抽象负责。当你在 DAG 中写schedule30 2 * * *时实际使用的是CronDataIntervalTimetable其定义位于 airflow-core/src/airflow/timetables/interval.pyclass CronDataIntervalTimetable(CronMixin, _DataIntervalTimetable): Timetable that schedules data intervals with a cron expression. This corresponds to schedulecron, 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. 这段类文档透露了两个关键事实schedulecron与CronDataIntervalTimetable一一对应支持五/六段 cron 表达式以及cron_presets预置值如daily。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夏令时转换在每个使用夏令时的时区每年发生两次转换方向墙钟行为术语春季进入 DST时钟前拨 1 小时某个小时被跳过例如 02:00 → 03:00gap / skip时间空洞秋季退出 DST时钟后拨 1 小时某个小时重复出现两次例如 02:00 → 01:00 再经历一次 01:00–02:00fold折返其中fold 是本次 bug 的重灾区。以源码注释中的例子airflow-core/src/airflow/timetables/_cron.py说明瑞士 2023 年在 UTC2 的凌晨 3 点退出 DST把时钟回拨到 UTC1 的凌晨 2 点。于是当地墙钟上会出现两个 02:00–03:00第一个 02:30 属于 UTC2DST 时段第二个 02:30 属于 UTC1标准时段即 fold 时段。如果 cron 表达式是30 * * * *每小时的第 30 分钟上一次运行在 02:30UTC2下一次本应落在 fold 后的 02:30UTC1——因为墙钟 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.pydef _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_timenaive) 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_timenaive) 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 * * * *之外的大多数表达式修复逻辑是把当前时刻current转成时间表时区的 naive 墙钟时间naive交给 croniter 求下一个/上一个调度点用make_aware(scheduled, self._timezone)把结果重新解释为该时区的带 fold 语义的时刻再经convert_to_utc转回 UTC 绝对时刻。这里make_aware对 fold 的处理是决定性的当 croniter 返回的 naive 时间恰好落在重复小时fold 区间内时make_aware会把fold1固定在第二次出现的位置上从而得到唯一的 UTC 映射。_get_prev中还额外加了一个防御循环由于make_aware可能把折返区间内的某个 tick 映射到当前或更晚的 UTC 时刻而不是更早代码会持续调用cron.get_prev()回退直到结果严格早于current为止——注释明确说明get_prev是严格递减的因此该循环必然终止。这一改动直接保证了上一个调度点语义在任何 DST 边界上都不会越界。每小时覆盖的 cronfold 时区问题的特判对于_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:30UTC2→ 下一次 02:30UTC1之间的墙钟间隔恰好是一个小时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.pyWhile 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 * * * *, timezoneEurope/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:00fold而非 DST在 fold 区间内间隔为[2023-10-29 01:00 UTC, 2023-10-29 01:30 UTC)仍是当地 02:00fold走出 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:00test_exiting_exact/test_exiting_fold退出 DST 时间隔终点落在第二次 02:00fold1对应的02:00 UTC上与 Airflow 历史行为保持一致测试注释明确说明There are two 2am local times on the 29th due to folding. We end on the second one (fold1). Theres no logical reason here; this is simply what Airflow has been doing since a long time ago, and theres 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全局默认在airflow.cfg中设置default_timezone Europe/Zurich或环境变量AIRFLOW__CORE__DEFAULT_TIMEZONEDAG 级覆盖在 DAG 构造时传入schedule30 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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考