如何修改Airflow DAG以指定过去日期运行API查询？

核心问题分析 #

你的DAG当前依赖date.today()生成查询日期，导致只能基于服务器实际当前日期运行。要实现针对过去日期的查询，有两种主流方案：利用Airflow内置的执行日期机制，或者通过自定义参数传入目标日期。

方案1：改用Airflow执行日期（execution_date）替代date.today() #

Airflow的每个DAG Run都有一个逻辑执行日期（execution_date），手动触发时可以自由指定这个日期，让DAG“假装”运行在该日期。修改代码如下：

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

# 假设你的default_args已定义
args = {
    'owner': 'airflow',
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

with DAG(
        'load_api_date',
        description='test',
        default_args=args,
        start_date=datetime(year=2023, month=4, day=17),
        schedule_interval=None,
) as dag:

    def get_data(**kwargs):
        # 从上下文获取逻辑执行日期，转为date对象
        exec_date = kwargs['execution_date'].date()
        # 计算delta_date：执行日期往前推2天
        delta_date = str(exec_date - timedelta(days=2))
        params = {"aggregation": "DAY", "date": delta_date, "returntype": "csv"}
        file_date = delta_date.replace('-', '')
        # 后续数据处理逻辑...

    run_task = PythonOperator(
        task_id='get_api_data',
        python_callable=get_data,
        provide_context=True,  # 必须开启，才能传递execution_date等上下文参数
    )

使用方式：
在Airflow UI中点击DAG的"Trigger DAG w/ config"，在弹出窗口中指定execution_date为目标日期（比如2023-04-18），DAG就会基于这个日期计算delta_date，相当于“假装”当前日期是18号。

方案2：添加自定义目标日期参数 #

如果需要更灵活的控制，可以允许手动触发时传入自定义的target_date参数，优先级高于执行日期：

def get_data(**kwargs):
    # 从DAG Run的配置中获取自定义参数
    dag_run_conf = kwargs.get('dag_run', {}).conf or {}
    target_date_str = dag_run_conf.get('target_date')
    
    if target_date_str:
        # 解析传入的目标日期
        target_date = datetime.strptime(target_date_str, '%Y-%m-%d').date()
    else:
        # 无传入参数时，使用执行日期
        target_date = kwargs['execution_date'].date()
    
    delta_date = str(target_date - timedelta(days=2))
    params = {"aggregation": "DAY", "date": delta_date, "returntype": "csv"}
    file_date = delta_date.replace('-', '')
    # 后续数据处理逻辑...

使用方式：
触发时在"Trigger DAG w/ config"的配置框中输入：

{"target_date": "2023-04-18"}

DAG就会基于这个目标日期计算delta_date。

关于start_date的作用 #

你的DAG设置了schedule_interval=None，属于手动触发型DAG，此时start_date仅作为Airflow要求的必填参数存在，对实际运行的日期逻辑没有任何影响。只有当DAG设置了自动调度的schedule_interval时，start_date才会影响自动生成的DAG Run的执行日期范围。

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