Airflow 2.0 Taskflow API中联合使用TaskGroup与BranchPythonOperator报错求解

你遇到的报错是因为BranchPythonOperator的返回值规则限制：

• BranchPythonOperator默认要求返回的是下游可执行任务的task_id，而你代码中get_tasks返回的'task_a'是TaskGroup的分组ID，不是具体任务的ID，Airflow无法识别分组ID作为分支跳转目标，因此抛出TaskNotFound错误。
• 额外注意：你的代码存在重复导入问题，from airflow.decorators import task和from airflow.operators.python import task重复导入了task装饰器，建议删除任意一个重复导入避免潜在冲突。

TaskGroup与BranchPythonOperator完全可以配合使用，下面提供两种通用兼容方案： #

方案1：返回TaskGroup内首个任务的完整ID #

TaskGroup内的所有任务，生成的实际task_id会自动带上{TaskGroupID}.前缀。你当前的task_a组内只有一个dummy_task，它的实际task_id为task_a.dummy_task，直接修改get_tasks的返回值即可：

def get_tasks(**kwargs):
    # 返回TaskGroup内首个任务的完整task_id
    return 'task_a.dummy_task'

如果你的TaskGroup内有多个任务，可在组内定义一个入口Dummy任务作为跳转目标：

with TaskGroup('task_a') as task_a:
    # 组内入口任务，作为分支跳转的目标
    group_entry = DummyOperator(task_id='group_entry')
    # 组内其他任务
    obj = dummy_task()
    other_task = dummy_task()
    # 定义组内依赖
    group_entry >> [obj, other_task]

# 分支函数直接返回入口任务的id即可
def get_tasks(**kwargs):
    return 'task_a.group_entry'

方案2：返回TaskGroup的下游任务ID列表 #

你也可以直接获取TaskGroup的所有下游任务ID返回，适合组内有多个并行入口任务的场景：

# 先在DAG作用域获取task_a的下游任务ID列表
group_task_ids = list(task_a.downstream_task_ids)

def get_tasks(**kwargs):
    # 返回整个组的入口任务ID列表
    return group_task_ids

修改完成后重新发布DAG触发运行：

• check_api任务会正常跳转到task_a组内的入口任务
• 整个task_a组会按定义的逻辑执行完成后进入final_step
• 另外两个分支task_b、task_c会被自动标记为skipped状态，符合分支逻辑

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