如何实现Airflow DAG每月第一个周五自动运行？

方法一：直接使用Cron表达式调度 #

Airflow的schedule_interval支持Cron表达式，要匹配每月第一个周五，可使用以下表达式：

schedule_interval = "0 0 1-7 * 5"

说明：该表达式表示每月1日至7日期间的周五，0点0分触发调度。由于每月第一个周五必然落在1-7号范围内（一周共7天），这个表达式可以精准满足需求。

如果你的业务使用非UTC时区，需在DAG的default_args中指定时区，避免时间偏差：

from datetime import datetime
import pytz

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2024, 1, 1, tzinfo=pytz.timezone('Asia/Shanghai')),
    'timezone': 'Asia/Shanghai'
}

方法二：基础调度+任务内校验（更严谨） #

若需更严谨的校验（比如避免极端场景下的误触发），可先设置每周五调度，再通过前置任务校验是否为当月第一个周五：

1. 设置DAG的基础调度为每周五：

schedule_interval = "0 0 * * 5"

2. 添加Python校验任务，判断执行日期是否为当月第一个周五，非目标日期则终止DAG：

from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

def check_first_friday(**context):
    execution_date = context['execution_date']
    # 计算当月第一个周五的日期
    first_day = execution_date.replace(day=1)
    # 周五对应的weekday是4（周一为0）
    delta_days = (4 - first_day.weekday()) % 7
    first_friday = first_day + timedelta(days=delta_days)
    
    if execution_date.date() != first_friday.date():
        raise ValueError("当前日期不是当月第一个周五，终止DAG运行")

# 创建校验任务
check_first_friday_task = PythonOperator(
    task_id='check_first_friday',
    python_callable=check_first_friday,
    provide_context=True,
    dag=dag
)

# 将校验任务设为业务任务的上游
check_first_friday_task >> your_business_task

只有当执行日期是当月第一个周五时，后续业务任务才会执行，否则DAG会因校验失败终止。

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