AirFlow中Backfill与Catchup的非冗余实用场景咨询

核心思路 #

要避免回填时任务重复执行无意义的全量操作，关键是让任务基于当前运行的逻辑日期（logical_date）/执行日期（execution_date）来处理对应时间窗口的数据。每个回填任务实例都会拿到专属的日期参数，只处理该日期的目标数据，而非重复处理全量内容。

实际案例：每日数据API拉取DAG #

假设我们需要拉取某API的每日用户行为数据，回填过去30天的数据，且每次任务只拉取对应日期的数据：

1. DAG定义（AirFlow 2.x版本） #

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

# 默认参数
default_args = {
    'owner': 'data_team',
    'retries': 1,
    'retry_delay': timedelta(minutes=5)
}

# 定义DAG，关闭自动追补（如需自动追补可设为True）
with DAG(
    dag_id='daily_api_data_pull',
    default_args=default_args,
    start_date=datetime(2024, 1, 1),
    schedule_interval='@daily',
    catchup=False,  # 禁止自动追补，如需手动回填用backfill命令
    tags=['data_pipeline']
) as dag:

    def pull_daily_data(**context):
        # 获取当前任务实例的逻辑日期（AirFlow 2.x推荐）
        execution_date = context['logical_date']
        # 格式化为API要求的日期格式（比如YYYY-MM-DD）
        target_date = execution_date.strftime('%Y-%m-%d')
        
        # 发起API请求，只拉取目标日期的数据
        api_url = f"https://api.example.com/user-behavior?date={target_date}"
        response = requests.get(api_url)
        response.raise_for_status()
        data = response.json()
        
        # 将数据写入本地文件或数据库（示例为写入文件）
        with open(f"/data/user_behavior_{target_date}.json", 'w') as f:
            json.dump(data, f)
        
        print(f"Successfully pulled data for {target_date}")

    # 定义任务
    pull_data_task = PythonOperator(
        task_id='pull_daily_data',
        python_callable=pull_daily_data,
        provide_context=True  # 允许任务获取DAG运行上下文
    )

    pull_data_task

2. 执行回填命令 #

运行以下命令回填2024年1月1日至1月30日的数据：

airflow dags backfill -s 2024-01-01 -e 2024-01-30 daily_api_data_pull

3. 效果说明 #

• 回填过程中，AirFlow会为每个日期（2024-01-01到2024-01-30）创建独立的任务实例。
• 每个任务实例都会使用对应的logical_date拉取当天的API数据，不会重复拉取全量数据，完全避免冗余。

关键注意事项 #

• 上下文参数传递：必须在PythonOperator中设置provide_context=True（AirFlow 2.x也可用op_kwargs传递日期），确保任务能获取到运行日期。
• 日期字段选择：AirFlow 2.x推荐使用logical_date而非execution_date，后者在调度逻辑上存在歧义（比如@daily的execution_date是前一天的日期）。
• Catchup的使用：如果开启catchup=True，DAG启动时会自动追补从start_date到当前日期的所有未运行任务，同样会基于每个日期执行对应任务，不会冗余。

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