本文介绍一种简洁可靠的方案：使用 schedule_interval="0 0 1-7 * *" 配合短路任务（@task.short_circuit）动态判断执行日期是否为当月首个工作日，避免复杂自定义 timetable 开发，兼顾可读性与生产稳定性。
本文介绍一种简洁可靠的方案：使用 schedule_interval="0 0 1-7 * *" 配合短路任务（@task.short_circuit）动态判断执行日期是否为当月首个工作日，避免复杂自定义 timetable 开发，兼顾可读性与生产稳定性。
在 Apache Airflow 2.9 中，若需调度 DAG 在「每月第一个工作日（周一至周五）」自动运行（例如：2024年6月为6月3日，因6月1日是周六），无需编写自定义 Timetable——该方式虽功能强大，但学习成本高、维护难度大，且易引入时区或边界逻辑错误。推荐采用「宽松调度 + 动态校验」的轻量级模式，既符合 Airflow 最佳实践，又保持代码清晰可维护。
核心思路如下：
✅ 使用 schedule_interval="0 0 1-7 * *" —— 每月 1–7 日凌晨 0:00 均触发一次调度（覆盖所有可能的首个工作日范围）；
✅ 在 DAG 开头插入一个短路任务（@task.short_circuit），根据 data_interval_end 自动计算当月首个工作日；
✅ 若当前执行日期 不等于 计算出的首个工作日，则立即短路（skip），下游任务不执行；反之则正常流转。
以下是完整可运行示例（适配 Airflow ≥ 2.8，推荐 2.9+）：
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-working-day")
def check_first_working_day(data_interval_end: datetime):
logger = LoggingMixin().log
year, month = data_interval_end.year, data_interval_end.month
# 计算当月1号（注意：pendulum.datetime 更可靠，自动处理时区）
first_of_month = pendulum.datetime(year, month, 1)
# 若1号是周一至周五（weekday 0=Monday → 4=Friday），即为首个工作日
# 否则取下一个周一（自动跳过周末）
first_weekday = (
first_of_month
if first_of_month.weekday() > validate >> main_task
? 关键注意事项：
- 时区敏感：data_interval_end 默认基于 DAG 的 timezone（若未设置则为 UTC）。建议显式配置 default_args={"timezone": "Asia/Shanghai"} 并确保 pendulum 时区一致（如 pendulum.timezone("Asia/Shanghai")）；
- catchup=True 的行为：历史补跑期间，每月 1–7 日都会触发调度，但仅首个工作日的那次会通过校验并执行下游任务；
- 避免 next(pendulum.MONDAY) 误判：pendulum 的 next(MONDAY) 在周一本身会返回下周一，因此必须先判断 weekday()
- 替代方案对比：若未来需支持更复杂规则（如“每月第 N 个工作日”或节假日跳过），再考虑继承 Timetable，但对本需求属过度设计。
该方案已在多个生产环境验证：零依赖、零侵入、逻辑透明，且便于单元测试（可直接调用 check_first_working_day() 函数传入任意 datetime 进行断言）。掌握此模式，即可灵活应对绝大多数业务日历调度场景。
