Airflow如何动态生成Dummy Operator并避免重复Task ID报错

错误原因 #

你遇到的DuplicateTaskIdFound错误由两个问题导致：

• 你已经手动预定义了task_id为step_2的Dummy任务，循环处理拆分后的子批次时，第一次循环会再次生成task_id为step_2的任务，直接触发ID重复
• 你在单个批次的子任务循环中重复调用Dummy生成逻辑，同一个批次内的多个任务会触发多次相同ID的Dummy创建，也会导致重复问题

优化实现方案 #

核心思路是用一个存储容器缓存已经创建的Dummy任务，避免重复生成，同时动态匹配批次依赖，完全不需要提前预定义批次对应的Dummy任务，新增任务对象只需更新列表即可自动适配：

from datetime import datetime, timedelta

from airflow import DAG
from airflow.contrib.operators.databricks_operator import \
    DatabricksRunNowOperator
from airflow.models import Variable
from airflow.operators.dummy_operator import DummyOperator

# These args will get passed on to each operator
default_args = {
    "owner": "airflow",
    "depends_on_past": False,
    "email_on_failure": False,
    "email_on_retry": False,
    "retries": 2,
    "retry_delay": timedelta(seconds=30),
    "is_paused_upon_creation": True,
    "timeout_seconds": 604800,
}

# Primary Control Block
with DAG(
    "name",
    start_date=datetime(2021, 1, 1),
    schedule_interval="@once",
    default_args=default_args,
    catchup=False,
    max_active_runs=1,
) as dag:

    # 固定前置任务
    imp_step_1 = DatabricksRunNowOperator(...)
    imp_step_2 = DatabricksRunNowOperator(...)

    # 第一批次无并发限制的任务列表
    data_obj_list_1 = ["a", "b", "c", "1", "2", "3"]
    # 后续需要限制并发的所有任务统一放在这个列表
    data_objs = ["d", "e", "f", "g", "h", "i", "x", "y", "z"]
    # 按每批次3个拆分
    data_objs = [data_objs[i : i + 3] for i in range(0, len(data_objs), 3)]

    def generate_tasks(job):
        # 注意task_id要包含job标识，避免重复
        return DatabricksRunNowOperator(
            task_id=f"databricks_task_{job}",
            # 其他你的原有参数
            ...
        )

    # 缓存已创建的Dummy任务，避免重复生成
    dummy_store = {}
    def get_or_create_dummy(step_num):
        if step_num not in dummy_store:
            dummy_store[step_num] = DummyOperator(
                task_id=f"step_{step_num}",
                trigger_rule="all_success",
            )
        return dummy_store[step_num]

    # 初始化前两个固定Dummy节点
    dummy_step_1 = get_or_create_dummy(1)
    dummy_step_2 = get_or_create_dummy(2)

    # 固定前置依赖
    imp_step_1 >> imp_step_2 >> dummy_step_1

    # 第一批次无并发限制的任务依赖
    for obj in data_obj_list_1:
        dummy_step_1 >> generate_tasks(obj) >> dummy_step_2

    # 动态处理后续有限流需求的批次
    for batch_idx, batch_objs in enumerate(data_objs):
        # 计算当前批次的上下游Dummy序号
        upstream_step = 2 + batch_idx
        downstream_step = upstream_step + 1
        upstream_dummy = get_or_create_dummy(upstream_step)
        downstream_dummy = get_or_create_dummy(downstream_step)
        # 给当前批次所有任务挂依赖
        for obj in batch_objs:
            upstream_dummy >> generate_tasks(obj) >> downstream_dummy

方案说明 #

• 所有Dummy任务通过get_or_create_dummy函数统一创建，重复调用只会返回已创建的实例，完全避免ID重复问题
• 后续新增任务只需往data_objs列表添加元素即可，无需手动修改Dummy任务定义、无需调整批次依赖，会自动拆分批次生成对应流程
• 完全保留原有DAG逻辑：前两个固定步骤串行，第一批次任务全量并行，后续批次最多3个任务并行，批次之间串行执行
• 生成的Databricks任务ID添加了job标识，避免不同任务ID重复

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