如何在Airflow DAG中配置时间规则，实现不同任务按指定时间运行

需求拆解 #

你需要的配置逻辑如下：

• DAG 全局每小时触发一次，每次都执行 importdata 导入任务
• 只有每日8点、18点的两个DAG运行批次，才会在导入完成后执行 sendReport 报表发送任务，其余批次跳过该任务

具体实现 #

1. DAG 全局调度配置 #

使用 cron 表达式 0 * * * * 配置 schedule_interval，实现每小时整点触发的效果。

2. 报表任务执行条件控制 #

通过 ShortCircuitOperator 做前置判断，仅当运行时的小时数为8或18时，才放行后续的报表发送任务，否则直接跳过。

完整示例代码 #

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

default_args = {
    "owner": "airflow",
    "start_date": datetime(2024, 1, 1),
    "retries": 1,
    "retry_delay": timedelta(minutes=5),
    "depends_on_past": False
}

with DAG(
    dag_id="hourly_import_daily_report",
    default_args=default_args,
    schedule_interval="0 * * * *",
    catchup=False
) as dag:

    # 导入文件任务
    def importdata_func():
        # 替换为实际的导入逻辑
        print("文件导入完成")

    importdata_task = PythonOperator(
        task_id="importdata",
        python_callable=importdata_func
    )

    # 执行时间判断任务
    def check_report_time(**context):
        execution_hour = context["execution_date"].hour
        return execution_hour in (8, 18)

    check_time_task = ShortCircuitOperator(
        task_id="check_report_time",
        python_callable=check_report_time,
        provide_context=True
    )

    # 发送报表任务
    def sendReport_func():
        # 替换为实际的报表发送逻辑
        print("报表发送完成")

    sendReport_task = PythonOperator(
        task_id="sendReport",
        python_callable=sendReport_func
    )

# 配置任务依赖
importdata_task >> check_time_task >> sendReport_task

注意事项 #

• 如果不需要单独的判断任务，也可以直接把时间校验逻辑写在 sendReport_func 内部，不符合条件直接返回即可，最终效果一致
• catchup 设为 False 可避免DAG启动时自动补跑历史未执行的任务，避免生成冗余运行记录

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