如何实现Apache Airflow幂等DAG：固定任务重跑的时间参数

要让任务重跑时使用首次执行的时间参数，核心是基于固定的DAG运行时间而非任务实时运行时间计算时间窗口。以下是两种可行方案：

方案一：使用Airflow执行日期（推荐） #

Airflow的execution_date是每个DAG运行的固定标识，即使任务重跑，该值也不会改变。我们可以利用它来生成稳定的时间窗口。

步骤1：配置DAG时区 #

确保DAG使用US/Eastern时区，这样execution_date会自动匹配目标时区：

from airflow import DAG
from datetime import timedelta
import pendulum

default_args = {
    'owner': 'airflow',
    'start_date': pendulum.datetime(2024, 1, 1, tz='US/Eastern'),
    'retries': 2,
    'retry_delay': timedelta(minutes=5),
}

with DAG(
    'ant_logs_dag',
    default_args=default_args,
    schedule_interval=timedelta(hours=1),
    timezone='US/Eastern',
    catchup=False,
) as dag:

步骤2：修改KubernetesPodOperator的环境变量 #

使用Jinja模板基于execution_date生成时间参数：

from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator

ant_get_logs = KubernetesPodOperator(
    env_vars={
        "startTime": "{{ (execution_date - macros.timedelta(hours=1)).strftime('%Y-%m-%d %H:%M:%S') }}",
        "endTime": "{{ execution_date.strftime('%Y-%m-%d %H:%M:%S') }}",
        "timeZone": 'US/Eastern',
        "session": 'none',
    },
    volumes=[volume],
    volume_mounts=[volume_mount],
    task_id='ant_get_logs',
    image='test:1.0.0',
    image_pull_policy='Always',
    in_cluster=True,
    namespace=namespace,
    name='kubepod_ant_get_logs',
    random_name_suffix=True,
    labels={'app': 'backend', 'env': 'dev'},
    reattach_on_restart=True,
    is_delete_operator_pod=True,
    get_logs=True,
    log_events_on_failure=True,
)

关键说明 #

• execution_date是DAG运行的固定时间点，重跑时保持不变
• 通过macros.timedelta计算1小时前的起始时间，与原逻辑一致
• 若DAG默认使用UTC时区，可转换时区后计算：

  "startTime": "{{ execution_date.in_timezone('US/Eastern').subtract(hours=1).strftime('%Y-%m-%d %H:%M:%S') }}",
  "endTime": "{{ execution_date.in_timezone('US/Eastern').strftime('%Y-%m-%d %H:%M:%S') }}",
  

方案二：使用XCom存储首次生成的时间窗口 #

若需严格基于任务首次启动时间生成窗口，可通过XCom存储时间参数，重跑时直接读取：

步骤1：添加生成时间窗口的前置任务 #

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

def generate_time_window(**context):
    eastern = pytz.timezone('US/Eastern')
    end_time = datetime.now(eastern)
    start_time = end_time - timedelta(hours=1)
    context['ti'].xcom_push(key='startTime', value=start_time.strftime('%Y-%m-%d %H:%M:%S'))
    context['ti'].xcom_push(key='endTime', value=end_time.strftime('%Y-%m-%d %H:%M:%S'))

generate_times = PythonOperator(
    task_id='generate_time_window',
    python_callable=generate_time_window,
    provide_context=True,
    retries=0,  # 禁止重跑，避免生成新时间
    dag=dag,
)

步骤2：修改KubernetesPodOperator读取XCom数据 #

ant_get_logs = KubernetesPodOperator(
    env_vars={
        "startTime": "{{ ti.xcom_pull(task_ids='generate_time_window', key='startTime') }}",
        "endTime": "{{ ti.xcom_pull(task_ids='generate_time_window', key='endTime') }}",
        "timeZone": 'US/Eastern',
        "session": 'none',
    },
    # 其他参数保持不变
)

# 设置任务依赖
generate_times >> ant_get_logs

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