如何为单个Airflow DAG配置满足90分钟周期且跳过9点的Cron调度？

首先明确你之前失败的核心原因：Cron的分钟字段仅支持0-59的数值范围，*/90这种写法超出范围，自然无法生效。针对你的需求（90分钟调度周期+跳过上午9点运行），这里提供几种靠谱的实现方案：

方法1：Airflow多调度表达式（推荐，Airflow 2.2+适用） #

Airflow从2.2版本开始支持schedule参数直接接收Cron表达式列表，你可以直接把已经验证好的两个表达式配置进去，完美匹配需求：

from airflow import DAG
from datetime import datetime

with DAG(
    dag_id="your_target_dag",
    schedule=[
        "30 1,4,7,10,13,16,19,22 * * *",
        "00 3,6,12,15,18,21,0 * * *"
    ],
    start_date=datetime(2024, 1, 1),
    catchup=False
) as dag:
    # 在这里定义你的任务逻辑
    pass

配置后DAG会严格按照这两个Cron的时间点触发，刚好实现90分钟间隔，且自动跳过了上午9点（00 9 * * *不在调度列表内）。

方法2：拆分两个DAG（兼容旧版本Airflow） #

如果你的Airflow版本低于2.2，不支持多调度表达式，可以把业务逻辑抽成公共函数，拆成两个独立DAG分别对应两个Cron：

# 抽离公共任务逻辑，避免重复代码
def core_task_logic():
    # 这里写你的核心业务代码
    pass

# 第一个DAG：对应30分的调度规则
with DAG(
    dag_id="your_dag_part1",
    schedule="30 1,4,7,10,13,16,19,22 * * *",
    start_date=datetime(2024, 1, 1),
    catchup=False
) as dag1:
    task1 = PythonOperator(
        task_id="core_task",
        python_callable=core_task_logic
    )

# 第二个DAG：对应00分的调度规则
with DAG(
    dag_id="your_dag_part2",
    schedule="00 3,6,12,15,18,21,0 * * *",
    start_date=datetime(2024, 1, 1),
    catchup=False
) as dag2:
    task2 = PythonOperator(
        task_id="core_task",
        python_callable=core_task_logic
    )

方法3：Timedelta调度+任务跳过逻辑 #

如果偏好固定时间间隔的写法（比如timedelta(minutes=90)），可以在任务中加入分支判断，当执行时间为上午9点整时跳过任务：

from airflow.operators.python import PythonOperator, BranchPythonOperator
from airflow.operators.dummy import DummyOperator
from datetime import datetime, timedelta

def check_exec_time(**context):
    exec_time = context["execution_date"]
    # 判断是否是上午9点整
    if exec_time.hour == 9 and exec_time.minute == 0:
        return "skip_task"
    else:
        return "run_task"

def your_business_logic():
    # 这里写你的业务代码
    pass

with DAG(
    dag_id="your_dag_timedelta",
    schedule_interval=timedelta(minutes=90),
    start_date=datetime(2024, 1, 1),
    catchup=False
) as dag:
    check_branch = BranchPythonOperator(
        task_id="check_exec_time",
        python_callable=check_exec_time,
        provide_context=True
    )

    run_task = PythonOperator(
        task_id="run_task",
        python_callable=your_business_logic
    )

    skip_task = DummyOperator(
        task_id="skip_task"
    )

    check_branch >> [run_task, skip_task]

这种方式无需手动维护Cron列表，但如果需要开启catchup，要确保跳过逻辑能正确处理历史时间点。

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