Airflow跨DAG复用任务的实现及任务重复执行问题解决

任务复用方式可行性说明 #

你采用的封装Operator_generator公共类生成Operator实例的复用方式完全可行，属于Airflow官方推荐的任务复用最佳实践，完全符合不允许使用subdag的限制要求。
该方案可避免多DAG同逻辑任务的代码冗余问题，后续任务逻辑迭代仅需修改公共类代码即可，无需逐个调整所有DAG的任务定义，相比subdag方案也不存在死锁、调度性能损耗等固有缺陷，非常适合仅参数存在差异的同逻辑任务复用场景。

任务重复执行根因 #

你定位的触发逻辑完全正确：删除DAG后重启再触发的操作流程会同时生成两个运行实例：

• 外部触发器主动调用生成1次手动运行实例，此时模板变量{{ds}}、{{prev_ds}}取触发当日的日期
• DAG重新加载后，调度器会扫描该DAG的历史运行记录，若当月调度周期的任务没有成功运行记录，且DAG默认开启了补跑（catchup=True），调度器会自动生成1次月调度补跑实例，此时模板变量的取值符合月度调度周期规则。

解决方案 #

临时修复方案（界面操作） #

你总结的操作流程可快速解决单次重复运行问题：

• 在Airflow Web UI关闭对应DAG的调度开关
• 点击DAG列表最右侧的红色删除按钮清除该DAG的所有历史记录、运行实例
• 刷新页面等待DAG被调度器重新加载到列表后，重新开启调度即可，此时调度器不会再重复补跑历史任务

永久优化方案（配置层面） #

可在DAG定义中显式关闭补跑开关，从根源避免删除DAG重新加载后的重复补跑问题，示例配置如下：

from airflow import DAG
from datetime import datetime

with DAG(
    dag_id="你的DAGID",
    schedule_interval="@monthly", # 你的月度调度配置
    start_date=datetime(2024, 1, 1), # 你的DAG起始日期
    catchup=False, # 关闭历史补跑开关
) as dag:
    # 实例化Operator_generator生成任务的逻辑

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