Airflow DAG月末前n天调度配置咨询：如何实现每月最后一天前3天执行

Airflow 原生 cron 表达式没有直接支持「每月倒数第N天」的语法，无法仅通过 cron 实现需求，以下是两种常用的可落地实现方案：

方案1：cron 触发 + 日期校验跳过（全版本通用） #

先通过 cron 配置每月25日到31日的固定时间触发DAG，再在任务最前增加日期校验逻辑，不符合要求的运行直接跳过，兼容性最高，所有Airflow版本都可以用。
示例代码如下：

from airflow import DAG
from airflow.operators.python import ShortCircuitOperator
from datetime import datetime, timedelta
from airflow.utils.dates import days_ago

# 校验当前日期是否为当月倒数第三天
def check_is_third_last_day(**context):
    execution_date = context['execution_date'].date()
    # 计算当月最后一天
    if execution_date.month == 12:
        next_month_first = datetime(execution_date.year + 1, 1, 1).date()
    else:
        next_month_first = datetime(execution_date.year, execution_date.month + 1, 1).date()
    last_day_of_month = next_month_first - timedelta(days=1)
    # 校验是否和最后一天间隔3天
    return (last_day_of_month - execution_date).days == 3

default_args = {
    'owner': 'airflow',
    'start_date': days_ago(1),
}

with DAG(
    'run_third_last_day_of_month',
    default_args=default_args,
    schedule_interval='0 0 25-31 * *', # 每日0点触发，仅在25-31号运行
    catchup=False
) as dag:
    # 日期校验节点，不符合要求则直接跳过后续所有任务
    check_date = ShortCircuitOperator(
        task_id='check_run_date',
        python_callable=check_is_third_last_day,
        provide_context=True,
    )

    # 你的实际业务任务节点，和校验节点做上下游关联即可
    # check_date >> your_business_task

如果需要调整提前天数，只需要修改(last_day_of_month - execution_date).days == 3里的数值，同时调整cron的触发起始日期即可。

方案2：自定义Timetable（Airflow 2.2+ 原生优雅实现） #

Airflow 2.2版本开始支持自定义Timetable调度规则，可以直接让DAG仅在符合要求的日期生成运行实例，不会产生多余的跳过记录，更符合原生设计逻辑。
示例代码如下：

from airflow.timetables.base import DagRunInfo, DataInterval, Timetable, TimetableRestriction
from datetime import timedelta
from pendulum import Date, DateTime

# 自定义每月倒数第三天调度规则
class ThirdLastDayTimetable(Timetable):
    def next_dagrun_info(
        self,
        *,
        last_automated_data_interval: DataInterval | None,
        restriction: TimetableRestriction,
    ) -> DagRunInfo | None:
        # 计算基准日期
        if last_automated_data_interval is None:
            current_date = restriction.earliest.date()
        else:
            current_date = last_automated_data_interval.end.date() + timedelta(days=1)
        
        while True:
            # 计算当月最后一天
            if current_date.month == 12:
                next_month_first = Date(current_date.year + 1, 1, 1)
            else:
                next_month_first = Date(current_date.year, current_date.month + 1, 1)
            last_day_of_month = next_month_first - timedelta(days=1)
            target_run_date = last_day_of_month - timedelta(days=3)
            # 找到下一个符合要求的运行日期
            if target_run_date >= current_date:
                run_dt = DateTime.combine(target_run_date, DateTime.min.time(), tz=restriction.earliest.tz)
                return DagRunInfo.interval(start=run_dt, end=run_dt + timedelta(days=1))
            # 不符合则跳转到下一个月计算
            current_date = next_month_first

DAG配置时直接指定自定义的Timetable即可，不需要额外加校验节点：

with DAG(
    'run_third_last_day_of_month_timetable',
    default_args=default_args,
    schedule=ThirdLastDayTimetable(),
    catchup=False
) as dag:
    # 直接写你的业务任务即可
    # your_business_task

注意事项 #

两种方案都会自动适配不同月份的天数，平年2月会在25号运行、闰年2月会在26号运行、3月会在28号运行，完全符合需求示例，无需手动调整配置。

内容的提问来源于stack exchange，提问作者Infinite