如何优雅地在每周Airflow任务中使用运行日期？

嘿，这个问题我太熟了——Airflow的execution_date设计确实容易让新手懵，毕竟它是调度周期的起始时间，不是任务实际触发运行的时间！我给你分不同Airflow版本说下具体的解决办法：

Airflow 2.x 及以上（推荐用这个！） #

Airflow 2.0之后引入了**数据区间（Data Interval）**的概念，专门解决了这个困惑。其中data_interval_end这个变量，正好就是你要的“周期结束/实际该触发运行的日期”。

就拿你说的周调度场景举例：如果你的DAG在2018-01-08T11:00:00触发运行，那：

• execution_date（现在官方也叫data_interval_start）是2018-01-01T11:00:00（也就是这周的起始）
• data_interval_end就是2018-01-08T11:00:00（完全对应你要的文件日期）

直接在任务里用这个变量就行，给你写个PythonOperator的示例：

from airflow.decorators import task
from airflow.models import DAG
from datetime import datetime

with DAG(
    dag_id="weekly_file_processor",
    schedule_interval="@weekly",
    start_date=datetime(2018, 1, 1),
    catchup=False
):
    @task
    def grab_and_process_file(**context):
        # 把日期转成你需要的格式
        target_date = context["data_interval_end"].strftime("%Y-%m-%dT%H:%M:%S")
        # 拼接文件名，比如你收到的是"weekly_data_2018-01-08T11:00:00.csv"
        file_name = f"weekly_data_{target_date}.csv"
        print(f"开始处理文件：{file_name}")
        # 这里写你的文件处理逻辑

    grab_and_process_file()

Airflow 1.x 版本（如果还没升级的话） #

要是你还在用Airflow 1.x，没有data_interval_end也没关系，直接用execution_date加上你的调度间隔来计算就行。比如周调度的间隔是7天，那：

from airflow.operators.python_operator import PythonOperator
from airflow.models import DAG
from datetime import datetime, timedelta

default_args = {
    "owner": "your_name",
    "start_date": datetime(2018, 1, 1)
}

with DAG(
    dag_id="weekly_file_processor_v1",
    schedule_interval="@weekly",
    default_args=default_args,
    catchup=False
):
    def process_file(**context):
        exec_date = context["execution_date"]
        # 加上7天得到周期结束的日期，也就是你要的运行日期
        target_date = (exec_date + timedelta(days=7)).strftime("%Y-%m-%dT%H:%M:%S")
        file_name = f"weekly_data_{target_date}.csv"
        print(f"开始处理文件：{file_name}")

    process_file_task = PythonOperator(
        task_id="process_file",
        python_callable=process_file,
        provide_context=True
    )

另外还有个备选方案：直接拿dag_run.start_date，这个是任务实际开始执行的时间。不过要注意，如果你的任务因为资源紧张延迟了，这个时间会和调度时间有偏差。如果你的文件名是严格对应调度日期的，还是优先用前面的方法哦。

举个用dag_run.start_date的小例子：

def process_file(**context):
    actual_run_date = context["dag_run"].start_date.strftime("%Y-%m-%dT%H:%M:%S")
    file_name = f"weekly_data_{actual_run_date}.csv"
    print(f"开始处理文件：{file_name}")

最后提一句：如果你的DAG开了catchup，这些方法也能正确对应每个补跑任务的调度周期结束日期，完全不用担心出错～

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