如何正确配置Airflow DAG上线，避免重复执行历史任务

针对你在Airflow 2.10.3（AWS MWAA）中遇到的问题——部署DAG后由其他团队在未来某天启用时，Airflow会触发旧系统已完成的历史区间任务，以下是几个实用的解决方案：

方案1：通过Airflow变量动态控制start_date（推荐） #

部署DAG时，将start_date设为一个极远的未来日期，同时通过Airflow变量来动态调整实际启动时间。当另一团队准备启用DAG时，只需修改变量值为启用当天的调度区间起始时间，再开启DAG即可。

代码示例：

from airflow.models import Variable
import pendulum

@dag(
    start_date=pendulum.parse(
        Variable.get("my_migrated_dag_start_date", default="2034-01-01T17:30:00"),
        timezone="Europe/London"
    ),
    schedule_interval="30 17 * * *",
    catchup=False,
    tags=["system_migration"]
)
def my_migrated_dag():
    # 此处定义你的业务任务
    ...

• 部署时变量默认值设为2034-01-01T17:30:00，确保DAG不会自动触发任何任务
• 启用时，将变量值修改为启用当天的调度时间（比如2024-05-20启用，就设为2024-05-20T17:30:00）
• catchup=False会确保Airflow只执行start_date之后的调度区间，完全跳过历史任务

方案2：用LatestOnlyOperator过滤历史任务 #

在DAG中添加LatestOnlyOperator，该算子会让后续任务仅在最新的调度区间运行，所有历史区间的任务都会被直接跳过。

代码示例：

from airflow.operators.latest_only import LatestOnlyOperator

@dag(
    start_date=pendulum.now("Europe/London").subtract(years=1),  # 设为过去较远日期
    schedule_interval="30 17 * * *",
    catchup=False,
)
def my_migrated_dag():
    # 仅允许最新调度区间执行后续任务
    latest_only = LatestOnlyOperator(task_id="skip_historical_runs")
    
    # 业务任务组
    extract = ...  # 你的抽取任务
    transform = ...  # 你的转换任务
    load = ...  # 你的加载任务
    
    latest_only >> extract >> transform >> load

• 如果启用DAG时已过当天调度时间（比如17:30之后启用），当天的任务会被跳过，直接执行第二天的调度任务
• 适合不需要精确控制启动日期，只需要从启用后首个调度时间开始执行的场景

方案3：手动修改Last Execution Date（无代码修改） #

如果不想修改DAG代码，可通过Airflow UI手动设置DAG的Last Execution Date：

1. 进入目标DAG的详情页面
2. 点击右上角的Edit按钮
3. 在Last Execution Date字段中，输入启用当天的调度区间起始时间（比如2024-05-20 17:30）
4. 保存后启用DAG

Airflow会以这个时间为起点，只执行后续的调度任务，不会触发之前的历史区间。

核心注意事项 #

• 无论用哪种方案，必须保持catchup=False，这是阻止Airflow自动回补历史任务的基础配置
• MWAA环境中，修改变量或DAG配置后，需要等待1-5分钟让DAG解析器刷新，配置才会生效
• 方案1的变量控制方式最适合跨团队协作场景，无需修改代码即可灵活调整启动时间

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