本文介绍一种简洁可靠的方案：使用 schedule_interval="0 0 1-7 * *" 配合日期校验任务，动态判断当前执行是否落在当月首个工作日（周一至周五），从而实现每月仅运行一次、且严格对齐业务定义的“第一个工作日”调度。
本文介绍一种简洁可靠的方案：使用 schedule_interval="0 0 1-7 * *" 配合日期校验任务，动态判断当前执行是否落在当月首个工作日（周一至周五），从而实现每月仅运行一次、且严格对齐业务定义的“第一个工作日”调度。
在 Apache Airflow 2.9 中，原生 Cron 表达式无法直接表达“每月第一个工作日”这一复杂语义（例如：1月1日是周一 → 触发；2月1日是周四 → 触发；6月1日是周六 → 推迟至6月3日周一触发）。Airflow 的自定义 Timetable 虽然功能强大，但开发、测试和维护成本较高，对多数团队而言并非必要选择。
更推荐采用轻量级、可读性强、易调试的“宽松调度 + 条件校验”模式：
- 设置一个覆盖范围合理的 Cron 表达式（如 "0 0 1-7 "），确保 Airflow 在每月前7天每天凌晨0点都触发一次 DAG 运行；
- 在 DAG 开头插入一个轻量级校验任务，精确计算当月第一个工作日（即周一至周五中最早的那个日期）；
- 使用 @task.short_circuit 装饰器，仅当 data_interval_end.date() 等于该工作日时才放行后续任务，否则自动短路（short-circuit），跳过所有下游操作。
以下是完整可运行示例：
from datetime import datetime
import pendulum
from airflow import DAG
from airflow.decorators import task
from airflow.operators.empty import EmptyOperator
from airflow.utils.log.logging_mixin import LoggingMixin
@task.short_circuit(task_id="check-first-workday")
def check_first_workday(data_interval_end: datetime):
logger = LoggingMixin().log
year = data_interval_end.year
month = data_interval_end.month
# 计算当月1号，并推导首个工作日
first_day = pendulum.datetime(year, month, 1)
if first_day.weekday() > validate >> main_task
✅ 关键说明与最佳实践：
- data_interval_end 是 Airflow 2.2+ 推荐使用的调度上下文参数（替代已弃用的 execution_date），它代表本次调度窗口的结束时间（即触发时刻），对每日调度而言即为当日 00:00:00；
- pendulum 比原生 datetime 更可靠地处理星期计算（.weekday() 返回 0–6，周一为 0）；
- catchup=True 对首次部署或逻辑验证至关重要——它将按 schedule 自动补跑历史月份，帮助你快速确认 2024-06-03、2024-12-02 等边界案例是否正确触发；
- 若需更高精度（如要求任务在工作日 08:00 执行），可将 schedule 改为 "0 8 1-7 * *"，并同步调整校验逻辑中的时间比较维度（仍以 .date() 为准）；
- 此方案完全兼容 Airflow 2.9+ 的 DAG 处理机制，无需注册自定义 Timetable，也无需修改核心代码，运维友好、升级安全。
通过这种“宽进严出”的设计，你既规避了 Timetable 的复杂性，又获得了比纯 Cron 更灵活、更贴近业务语义的调度能力——真正实现「每月只跑一次，且永远准时落在第一个工作日」。
