如何配置Airflow DAG每两月在最后一个周日凌晨3点运行？

一、原有Cron表达式的问题 #

你之前使用的0 3 * */2 7会匹配所有偶数月的每个周日，而不是仅最后一个。因为*/2指定偶数月，7指定周日，两者是逻辑“且”的关系，所以每个偶数月的周日都会触发调度，不符合需求。

纯标准Cron语法无法直接表达“偶数月最后一个周日”，因为Cron的日期和星期字段是逻辑与关系，无法直接筛选“当月最后一个星期X”并限定月份。不过可以结合Airflow特性或扩展Cron语法实现需求。

二、可行解决方案 #

方案1：使用扩展Cron表达式（依赖croniter） #

Airflow底层使用的croniter库支持扩展Cron语法，用L结合星期字段可以表示当月最后一个指定星期几。针对需求可以使用：

0 3 L 2,4,6,8,10,12 7

这个表达式的含义是：偶数月（2/4/6/8/10/12月）的最后一个周日凌晨3点。可以用以下代码本地验证：

from croniter import croniter
from datetime import datetime

cron = croniter('0 3 L 2,4,6,8,10,12 7', datetime(2024,1,1))
for _ in range(5):
    print(cron.get_next(datetime))

输出会是2024年2月25日、4月28日、6月30日等符合要求的日期。

方案2：结合分支任务做日期判断 #

如果担心扩展Cron语法的兼容性，可以先匹配偶数月的所有周日，再通过分支任务筛选出最后一个周日：

1. 设置基础调度规则：

dag = DAG(
    dag_id=DAG_ID,
    default_args=DEFAULT_ARGS,
    schedule_interval='0 3 * 2,4,6,8,10,12 7',
    start_date=datetime(2021, 6, 1),
    max_active_runs=1
)

2. 添加日期判断分支任务：

from airflow.operators.python import BranchPythonOperator
from airflow.operators.dummy import DummyOperator
import pendulum

def is_last_sunday_of_month(**context):
    execution_date = pendulum.parse(context['execution_date'].isoformat())
    next_sunday = execution_date.add(days=7)
    # 判断下一个周日是否超出当前月份
    return 'execute_task' if next_sunday.month != execution_date.month else 'skip_task'

branch_task = BranchPythonOperator(
    task_id='check_last_sunday',
    python_callable=is_last_sunday_of_month,
    provide_context=True,
    dag=dag
)

execute_task = DummyOperator(task_id='execute_task', dag=dag)
skip_task = DummyOperator(task_id='skip_task', dag=dag)

branch_task >> [execute_task, skip_task]
# 实际业务任务挂载到execute_task节点后

方案3：自定义Timetable（Airflow 2.2+） #

Airflow 2.2及以上版本支持自定义Timetable，这是最灵活的调度控制方式：

from airflow.timetables.base import DagRunInfo, DataInterval
from airflow.timetables.interval import CronDataIntervalTimetable
from datetime import datetime
import pendulum

class EvenMonthLastSundayTimetable(CronDataIntervalTimetable):
    def __init__(self):
        # 基础规则：偶数月的所有周日凌晨3点
        super().__init__('0 3 * 2,4,6,8,10,12 7', timezone=pendulum.timezone('UTC'))
    
    def next_dagrun_info(
        self,
        last_automated_dagrun: datetime | None,
        restriction: DagRunInfo.Restriction,
    ) -> DagRunInfo | None:
        candidate = super().next_dagrun_info(last_automated_dagrun, restriction)
        if not candidate:
            return None
        
        execution_date = candidate.data_interval.end
        next_sunday = execution_date.add(days=7)
        # 循环筛选，直到找到当月最后一个周日
        while next_sunday.month == execution_date.month:
            candidate = super().next_dagrun_info(execution_date, restriction)
            if not candidate:
                return None
            execution_date = candidate.data_interval.end
            next_sunday = execution_date.add(days=7)
        return candidate

# DAG中使用自定义Timetable
dag = DAG(
    dag_id=DAG_ID,
    default_args=DEFAULT_ARGS,
    timetable=EvenMonthLastSundayTimetable(),
    start_date=datetime(2021, 6, 1),
    max_active_runs=1
)

总结 #

• 追求简洁优先尝试扩展Cron表达式0 3 L 2,4,6,8,10,12 7；
• 需兼容性保障选分支任务判断的方式；
• Airflow 2.2+版本推荐自定义Timetable，灵活性最高。

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