如何使Airflow新DAG的历史运行实例默认置为成功状态？

可以实现，且无需通过UI手动逐个设置历史实例状态，以下是两种可行方案：

方案一：利用Airflow内置参数（Airflow 2.2及以上版本适用） #

在DAG定义时，同时配置两个关键参数：

• catchup=False：阻止Airflow自动补跑从start_date到部署日之间的历史任务
• initial_state='success'：让Airflow在DAG首次部署时，直接将start_date到当前日期范围内所有应调度的DAG Run标记为成功状态

示例代码片段：

from airflow import DAG
from datetime import datetime

with DAG(
    dag_id='monthly_historical_success',
    start_date=datetime(2023, 1, 1),
    schedule_interval='@monthly',
    catchup=False,
    initial_state='success'
) as dag:
    # 后续任务定义
    pass

注意：initial_state仅在DAG首次创建时生效，后续修改DAG不会影响已生成的DAG Run状态。

方案二：任务分支判断（兼容全版本Airflow） #

如果你的Airflow版本低于2.2，可在DAG起始位置添加分支判断逻辑：

• 检查当前任务的执行日期是否早于部署日期（2023-10-01）
• 若早于部署日期，直接跳过所有后续任务并标记为成功；若为部署日及之后的日期，则正常执行任务

示例代码片段：

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

def check_execution_date(**context):
    exec_date = context['execution_date']
    deploy_date = datetime(2023, 10, 1)
    if exec_date < deploy_date:
        return 'mark_as_success'
    else:
        return 'normal_task'

def mark_success(**context):
    # 无需执行实际逻辑，直接标记任务成功
    pass

with DAG(
    dag_id='monthly_historical_success',
    start_date=datetime(2023, 1, 1),
    schedule_interval='@monthly',
    catchup=True  # 允许补跑，但通过分支逻辑跳过实际执行
) as dag:
    check_branch = BranchPythonOperator(
        task_id='check_execution_date',
        python_callable=check_execution_date,
        provide_context=True
    )

    mark_success_task = PythonOperator(
        task_id='mark_as_success',
        python_callable=mark_success
    )

    # 定义正常执行的任务
    normal_task = PythonOperator(
        task_id='normal_task',
        python_callable=lambda: print("正常执行月度任务")
    )

    check_branch >> [mark_success_task, normal_task]

这种方式下，历史周期的任务会自动进入mark_as_success分支，无需实际执行即可标记为成功，部署日之后的任务则正常运行。

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