Airflow：如何实现每月初运行DAG并将逻辑日期设为当月起始日

这个需求完全可以搞定，而且有两种实用的方案——用Airflow内置配置快速实现，或者用自定义时间表满足更复杂的场景，我给你详细说清楚：

方案1：用内置Cron表达式快速实现 #

Airflow默认是周期结束后再执行，但我们可以通过调整schedule_interval、start_date和catchup参数，让DAG在当月初执行，同时把逻辑日期设为当月第一天。

具体步骤：

• 把schedule_interval设为0 0 1 * *，也就是每月1日0点触发执行
• start_date设为你想要启动的第一个月的第一天，比如datetime(2024, 3, 1)
• 一定要开启catchup=False，避免Airflow回溯执行之前的历史周期
• 按需配置execution_timeout等参数，防止任务超时

直接上代码示例：

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

def monthly_task_logic():
    print(f"执行每月任务，逻辑日期为: {{{{ ds }}}}")  # 这里会输出当月第一天

with DAG(
    dag_id='monthly_start_of_month_dag',
    start_date=datetime(2024, 3, 1),
    schedule_interval='0 0 1 * *',
    catchup=False,
    tags=['monthly', 'start_of_month']
) as dag:
    run_monthly_task = PythonOperator(
        task_id='run_monthly_logic',
        python_callable=monthly_task_logic
    )

这段代码里，3月1日执行时，DAG的逻辑日期ds就是2024-03-01，完全符合你的要求。因为我们指定了每月1日执行，start_date是当月第一天，加上catchup=False，每次执行的逻辑日期都会和执行当天所在月的起始日对齐。

方案2：自定义时间表（Custom Timetable） #

如果你的需求更复杂——比如需要跳过节假日调整执行时间，或者有特殊的周期计算逻辑，自定义时间表就是更好的选择。Airflow 2.2及以上版本支持自定义时间表，能完全掌控逻辑日期和实际执行时间的映射关系。

下面是一个自定义每月初时间表的实现：

from airflow.timetables.base import DagRunInfo, DataInterval, TimeRestriction
from airflow.timetables.interval import CronTimetable
from datetime import datetime, timedelta
from pendulum import DateTime

class MonthlyStartTimetable(CronTimetable):
    def __init__(self, timezone="UTC"):
        # 基础Cron规则：每月1日0点执行
        super().__init__("0 0 1 * *", timezone=timezone)

    def next_dagrun_info(
        self,
        last_automated_dagrun: DateTime | None,
        restriction: TimeRestriction,
    ) -> DagRunInfo | None:
        # 获取基础Cron计算出的下一次执行时间
        base_info = super().next_dagrun_info(last_automated_dagrun, restriction)
        if not base_info:
            return None
        
        # 将逻辑日期（data_interval.start）设为当月第一天
        execution_date = base_info.data_interval.end
        start_of_month = execution_date.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
        # 计算周期结束时间（当月最后一刻）
        end_of_month = (start_of_month + timedelta(days=32)).replace(day=1) - timedelta(microseconds=1)
        
        return DagRunInfo(
            data_interval=DataInterval(start=start_of_month, end=end_of_month),
            run_after=execution_date,
        )

然后在DAG中使用这个自定义时间表：

with DAG(
    dag_id='custom_monthly_start_dag',
    start_date=datetime(2024, 3, 1),
    timetable=MonthlyStartTimetable(timezone="Asia/Shanghai"),  # 按需指定时区
    catchup=False,
    tags=['custom_timetable', 'monthly']
) as dag:
    # 这里定义你的任务，比如PythonOperator、BashOperator等
    pass

这个自定义时间表的核心逻辑是：

• 基于内置的CronTimetable实现，保证每月1日触发执行
• 手动将data_interval.start设置为当月第一天，作为DAG的逻辑日期
• run_after保持为每月1日，确保实际执行时间是当月初

几个关键注意事项 #

• 时区问题：如果你的业务用的是非UTC时区，一定要在DAG或自定义时间表中指定正确的timezone参数，避免执行时间出现偏移
• catchup设置：如果开启catchup=True，Airflow会自动补全从start_date到当前日期的所有周期，根据你的需求建议关闭
• 本地测试：可以用dag.test()方法在本地测试DAG的执行时间和逻辑日期，提前验证是否符合预期

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