Airflow动态任务映射：如何为每个映射任务配置专属后续任务

AI 小北

2026-7-8

解决方案

要实现每个映射后的Task B实例对应独立的Task C实例（即B[0]→C[0]、B[1]→C[1]各自独立执行，无需等待所有Task B完成），核心是让Task C的映射关联对应Task B实例的输出，而非直接依赖原始的Task A输出。

正确写法示例

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

@dag(start_date=datetime(2024, 1, 1), schedule=None)
def mapped_task_dag():
    @task
    def taskA():
        # 返回用于映射的数据源，比如3个元素的列表
        return [{"v": "item1"}, {"v": "item2"}, {"v": "item3"}]

    @task
    def taskB(v):
        print(f"Processing {v} in Task B")
        return f"processed_{v}"

    @task
    def taskC(v):
        print(f"Processing {v} in Task C")

    # 第一步：映射生成Task B的所有实例
    mapped_b = taskB.expand_kwargs(v=taskA())
    # 第二步：基于每个Task B实例的输出，映射生成对应的Task C实例
    mapped_c = taskC.expand(v=mapped_b)

mapped_task_dag()

为什么之前的写法不生效

第一种写法：taskB.expand_kwargs(v=taskA) >> taskC.expand(v=taskA)
这里Task C的映射直接依赖Task A的输出，和Task B的实例没有关联。Airflow会将所有Task B实例视为一个整体，所有Task C实例也视为一个整体，因此必须等所有Task B完成后才会启动Task C。
第二种写法：(Task B >> TaskC).expand_kwargs(Task A)
语法错误，expand_kwargs是Task实例的方法，不能直接作用于Task之间的依赖链（>>返回的是Dependency对象），因此无法编译通过。