Airflow 2 @task松散耦合场景下XComArg返回值单属性传递异常问题

问题根因 #

默认情况下Airflow会把@task修饰的函数返回值整体作为单个XCom存储，此时直接对XComArg对象执行下标取值（如res['a']）不会被解析为字典字段提取操作，因此下游任务拿到的参数值为None。

推荐解决方案（完全满足松耦合要求） #

给上游任务的@task装饰器添加multiple_outputs=True参数，开启多输出模式。开启后返回字典的每个key都会被独立存储为一个XCom条目，直接传递res['a']即可让下游拿到对应字段的正确值，下游任务完全不需要感知参数来源，也不需要处理上游返回的完整字典。

修正后代码 #

@dag(...)
def _dag():
    @task(multiple_outputs=True)
    def A(**ctx):
        return {'a': 42, 'b': 'B', 'c': 'C'}

    @task
    def B(a, **ctx):
        print('A', a) # 运行时输出为 A 42

    res = A()
    B(a=res['a'])

dag = _dag()

低版本兼容方案 #

如果使用的Airflow版本低于2.1（不支持multiple_outputs特性），可以用模板语法手动提取对应字段：

B(a="{{ ti.xcom_pull(task_ids='A')['a'] }}")

该方案需要硬编码上游任务ID，耦合度更高，仅作为降级方案使用。

内容的提问来源于stack exchange，提问作者rodrigo-silveira