如何配置Airflow DAG实现每日运行且包含当日执行？代码异常问题求助

嘿，我来帮你梳理下这个问题！你遇到的其实是Airflow新手常踩的两个坑：一个是Airflow核心的调度时间逻辑，另一个是时区不一致导致的显示问题。咱们一步步来解决：

问题根源拆解 #

1. Airflow的调度时间逻辑：
   Airflow的execution_date（Airflow 2.x里也叫logical_date）代表的是这个DAG运行对应的时间周期起点，而实际触发运行的时间是这个周期的结束点。比如你设置schedule_interval='30 17 * * *'，意思是“每天17:30触发一次运行”，但这个运行对应的logical_date是前一天的17:30——因为它对应的是“昨天17:30到今天17:30”这个时间周期。这就是为什么你把start_date设为昨天，界面会显示logical_date是昨天的原因。

2. 时区不一致导致的显示问题：
   你在DAG里用了Europe/Madrid时区，但如果你的Airflow集群默认时区是UTC（默认配置），web界面上显示的时间会转换成UTC时间。比如马德里时间12月6日17:30对应UTC时间16:30，看起来还是同一天，但如果是更早的时间点，就会显示成前一天的日期，让你误以为执行日期错误。

解决方案（分两种场景） #

场景1：只要任务处理当天数据，不在乎界面显示的logical_date #

这种情况最简单，不用改变调度逻辑，只需要在任务里直接获取运行当天的马德里时区日期，不管logical_date是什么：

修改后的代码：

import pendulum
from airflow.models.dag import DAG
from airflow.operators.empty import EmptyOperator
from airflow.operators.python import PythonOperator  # Airflow 2.x推荐写法，旧路径已弃用

with DAG(
    dag_id="example",
    schedule_interval='30 17 * * *',
    start_date=pendulum.datetime(2024, 12, 5, 17, 20, tz="Europe/Madrid"),  # 设为前一天的同一时间
    catchup=False,
    tags=["example3"],
) as dag:

    def print_hello():
        # 直接获取马德里时区的当前日期
        today = pendulum.now("Europe/Madrid").date()
        print(f'hello, today is {today}')
        print('hello')

    task1 = EmptyOperator(task_id="task1")
    task2 = EmptyOperator(task_id="task2")
    python_task = PythonOperator(
        task_id='my_python_task',
        python_callable=print_hello,
        op_kwargs={'key': 'value'}
    )
    task1 >> task2 >> python_task

修改说明：

• 把start_date设为前一天的17:20：这样Airflow会在当天17:30触发第一次运行，不会漏掉当日的执行
• 任务里用pendulum.now()直接获取当天日期：保证任务处理的是运行当天的数据，不受logical_date的影响
• 替换了PythonOperator的导入路径：适配Airflow 2.x的最新写法，避免弃用警告

场景2：既要处理当天数据，又要界面显示的logical_date是当天 #

这种情况需要调整Airflow的时区配置，同时优化DAG的时间设置：

1. 修改Airflow集群的时区配置：
   在airflow.cfg中修改以下两个参数，保存后重启Airflow服务：

   default_timezone = Europe/Madrid
   webserver_timezone = Europe/Madrid
   

   这样整个Airflow系统的时间都会用马德里时区，界面显示的logical_date和实际运行时间会一致。

2. 调整DAG的schedule_interval和start_date：
   如果希望logical_date是当天的日期，可以把schedule_interval设为@daily，然后设置start_date为当天的00:00，这样每天00:00触发的任务，logical_date就是当天的日期。不过如果你需要固定17:30运行，还是用原来的CRON表达式，配合时区配置即可——此时界面显示的logical_date会是马德里时区的前一天17:30，但你可以在任务名称或日志里标注当天的日期，方便查看。

额外的小建议 #

如果你用的是Airflow 2.x，建议使用@dag装饰器的写法，代码更简洁：

import pendulum
from airflow.decorators import dag, task
from airflow.operators.empty import EmptyOperator

@dag(
    dag_id="example",
    schedule_interval='30 17 * * *',
    start_date=pendulum.datetime(2024, 12, 5, 17, 20, tz="Europe/Madrid"),
    catchup=False,
    tags=["example3"],
)
def example_dag():
    task1 = EmptyOperator(task_id="task1")
    task2 = EmptyOperator(task_id="task2")

    @task(task_id='my_python_task')
    def print_hello():
        today = pendulum.now("Europe/Madrid").date()
        print(f'hello, today is {today}')

    task1 >> task2 >> print_hello()

example_dag()

备注：内容来源于stack exchange，提问作者Com Media