Airflow 2.10中动态任务映射与BranchPythonOperator的兼容性问题及TaskFlow API分支实现示例

咱先唠唠大家在Airflow 2.10里常碰到的一个小麻烦：用动态任务映射的时候，老款的BranchPythonOperator经常会出现兼容问题，比如动态映射后分支判断跑偏、后续任务接不上之类的情况。其实换个思路，用TaskFlow API来写分支逻辑，不仅写法更清爽，还能完美适配动态任务映射的场景，我给你整个最简实现的例子，一看就明白～

用TaskFlow API实现分支操作的最简示例 #

这个DAG的逻辑特别直白：根据branch_on_condition任务返回的结果，要么执行odd_task要么even_task，而且这两个分支任务还会用到分支判断时依赖的return_int返回值，最后不管走哪个分支，都会执行final_task收尾。

from airflow.decorators import dag, task
from datetime import datetime

@dag(start_date=datetime(2024, 1, 1), schedule=None, catchup=False)
def branch_taskflow_dag():
    @task
    def return_int():
        # 这里可以替换成动态任务映射生成的变量，适配你的动态场景
        return 3

    @task.branch
    def branch_on_condition(num):
        # 根据输入整数的奇偶性决定分支走向
        if num % 2 == 1:
            return "odd_task"
        else:
            return "even_task"

    @task
    def odd_task(num):
        print(f"正在执行奇数任务，输入的数值是: {num}")
        return f"奇数任务处理完成，数值为: {num}"

    @task
    def even_task(num):
        print(f"正在执行偶数任务，输入的数值是: {num}")
        return f"偶数任务处理完成，数值为: {num}"

    @task
    def final_task(processed_result):
        print(f"最终任务收到处理结果: {processed_result}")

    # 编排任务依赖关系
    num = return_int()
    branch_decision = branch_on_condition(num)
    odd_result = odd_task(num)
    even_result = even_task(num)
    # 分支任务指向对应分支，最后统一走到最终任务
    branch_decision >> [odd_result, even_result] >> final_task(odd_result if num %2 ==1 else even_result)

# 实例化DAG
branch_dag = branch_taskflow_dag()

再跟你拆解下关键细节：

• return_int任务可以返回固定值，也能直接换成动态任务映射出来的参数，完美适配动态场景
• 用@task.branch装饰器直接把普通任务变成分支任务，返回要执行的下一个任务ID就行，比老款BranchPythonOperator的写法清爽太多
• 任务依赖关系用>>直接串联，逻辑一目了然，不会出现动态映射时的衔接故障

至于动态任务映射和TaskFlow分支的兼容性，完全不用发愁，你只需要在分支任务里接收映射过来的每一个参数，返回对应分支的任务ID，Airflow会自动处理好映射后的任务实例和分支的对应关系，比用BranchPythonOperator省心多了。

备注：内容来源于stack exchange，提问作者GazYah