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为当前环境时间文中所有涉及日期、版本的陈述均以本仓库代码与文档为唯一依据。目录从需求说起为什么需要 TimetableTimetable 核心接口与基础类型Timetable 注册与 DAG 接入实现调度逻辑两个核心方法参数化 Timetable 与序列化Timetable 在 UI 中的展示summary 与 description自定义 generate_run_id测试验证以 test_workday_timetable.py 为例延伸阅读从需求说起为什么需要 TimetableAirflow 内置的 cron 表达式调度能力非常强大但无法覆盖所有业务场景。本文以一个典型需求为例一家公司希望在每个工作日结束后处理当天工作时间内采集的数据。直觉上大家首先会想到 cron 表达式schedule0 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 及其任务对调度时间的所有限制包含三个字段earliestDAG 可以被调度的最早时间由 DAG 和所有任务的start_date参数计算得出如果没有start_date则为Nonelatest与earliest类似由end_date参数计算得出catchup布尔值反映 DAG 的catchup参数默认为FalseDagRunInfo一个NamedTuple描述一次 DagRun 的调度信息包含run_afterDagRun 最早可以被创建并调度任务的时间和data_interval两个字段。其interval(start, end)类方法用于构造区间结束即触发的运行且data_interval.end run_after恒成立Timetable一个Protocol是所有 Timetable 类需要实现的接口其中最重要的两个方法是next_dagrun_info和infer_manual_data_interval。# DagRunInfo 的两种构造方式 info DagRunInfo( data_intervalDataInterval(startstart, endend), run_afterrun_after, ) # 通常我们希望在区间结束时立即触发因此有更简洁的快捷方式 info DagRunInfo.interval(startstart, endend) 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 Fridayperiodic: bool True该 Timetable 是否周期性运行scheduleNone和once等特殊设置会将其置为Falsecan_be_scheduled: bool True能否真正以自动化方式调度运行NullTimetable会将其置为Falserun_ordering该 Timetable 触发的运行在 UI 中的排序字段默认为(data_interval_end, logical_date)active_runs_limit: int | None NoneDAG 同时可处于活跃状态的最大运行数在 DAG 初始化时调用返回值用作 DAG 的默认max_active_runssummary属性用于在 Web UI 中展示 Timetable 的简短摘要默认实现返回类的类型名type_name属性主要用于按 Timetable 类型过滤 DAG内置 Timetable 返回类名自定义 Timetable 返回完整导入路径validate()方法在 DAG 放入 dagbag 时进行运行时校验失败时抛出AirflowTimetableInvalidserialize()/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 文件中使用这个 Timetableimport pendulum from airflow.sdk import DAG from airflow.example_dags.plugins.workday import AfterWorkdayTimetable with DAG( dag_idexample_after_workday_timetable_dag, start_datependulum.datetime(2021, 3, 10, tzUTC), scheduleAfterWorkdayTimetable(), tags[example, timetable], ): ...仓库中 workday.py 的WorkdayTimetablePlugin正是通过timetables [AfterWorkdayTimetable]这种方式把自定义 Timetable 注册进 Airflow 的插件系统。实现调度逻辑两个核心方法当 Airflow 的 scheduler 遇到一个 DAG 时会调用以下两个方法之一来决定何时调度该 DAG 的下一次运行next_dagrun_infoscheduler 用它来了解 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(days1)).date(), Time.min).replace(tzinfoUTC) # Skip backwards over weekends and holidays to find last run start self.get_next_workday(start, incr-1) return DataInterval(startstart, end(start timedelta(days1))) # [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(days1)).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. Dont schedule. elif not restriction.catchup: # If the DAG has catchupFalse, today is the earliest to consider. next_start max(earliest, DateTime.combine(Date.today(), Time.min, tzinfoUTC)) elif earliest.time() ! Time.min: # If earliest does not fall on midnight, skip to the next day. next_start DateTime.combine(earliest.date() timedelta(days1), Time.min) else: next_start earliest # Skip weekends and holidays next_start self.get_next_workday(next_start.replace(tzinfoUTC)) if restriction.latest is not None and next_start restriction.latest: return None # Over the DAGs scheduled end; dont schedule. return DagRunInfo.interval(startnext_start, end(next_start timedelta(days1))) # [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.pydef get_next_workday(self, d: DateTime, incr1) - 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(startnext_start, endnext_start).to_pydatetime() if next_start not in holidays: break next_start next_start.add(daysincr) return next_start它通过循环递增/递减天数跳过周六weekday5、周日weekday6以及节假日找到下一个或上一个当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, incr1) - 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_idexample_workday_timetable, start_datependulum.datetime(2021, 1, 1, tzUTC), scheduleAfterWorkdayTimetable(), tags[example, timetable], ): EmptyOperator(task_idrun_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(days1) return DagRunInfo( data_intervalDataInterval(startstart, endend), run_afterDateTime.combine(end.date(), self._schedule_at).replace(tzinfoUTC), )如果要把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 fafter each workday, at {self._schedule_at}于是对于如下声明的 DAGwith DAG( scheduleSometimeAfterWorkdayTimetable(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__会用ExpressionDescriptorcron-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_typerun_type, logical_datelogical_date, data_intervaldata_interval, **extra )注意run_id长度限制为250 个字符同一个 DAG 内的run_id必须唯一。Timetable协议中generate_run_id的默认实现base.py是run_type.generate_run_id(suffixrun_after.isoformat())。除generate_run_id外接口还提供next_dagrun_info_v2next_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, day1天)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交互构成了一个完整的文档 → 示例实现 → 测试验证闭环。延伸阅读公开接口的完整说明airflow.timetables.base模块airflow-core/src/airflow/timetables/base.py对子类需要实现的方法有详尽注释内置 Timetable 实现cron 表达式与时间差驱动的数据区间 Timetable 在 airflow-core/src/airflow/timetables/interval.py触发器类 Timetable 在 airflow-core/src/airflow/timetables/trigger.pyscheduleNone、once等平凡 Timetable 在 airflow-core/src/airflow/timetables/simple.py插件注册机制见 插件文档关于 Airflow 调度概念的更多背景可参考 调度与定时 相关文档。【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考