Airflow动态任务映射报错：无法使用自定义XCom Key执行expand映射

问题原因 #

当你尝试用expand()直接引用任务的自定义XCom Key（比如nicks）时，Airflow默认的任务输出引用（如get_name.output）指向的是任务的默认XCom（key为return_value），若未正确指定要获取的自定义XCom Key，就会触发ValueError。

解决方案 #

根据你的任务实现方式，分两种情况处理：

情况1：get_name任务通过return返回包含nicks的字典 #

如果get_name任务直接return包含目标列表的字典（默认XCom存储的就是这个字典），无需额外指定自定义XCom Key，直接从返回的字典中提取nicks列表即可：

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

@dag(start_date=datetime(2024,1,1), schedule=None, catchup=False)
def dynamic_map_dag():
    @task
    def get_name():
        # 直接返回包含nicks的字典，默认XCom key为'return_value'
        return {"name": "Jerald", "nicks": ["Jerry", "J-Rock", "J-Man"]}

    @task
    def process_nick(nick):
        print(f"Processing nickname: {nick}")
        return f"Processed {nick}"

    name_data = get_name()
    # 从返回的字典中提取nicks列表，传给expand实现动态映射
    process_nick.expand(nick=name_data["nicks"])

dynamic_map_dag()

情况2：get_name任务使用自定义XCom Key存储nicks列表 #

如果get_name任务是通过ti.xcom_push主动推送了自定义XCom，需要用XComArg显式指定要获取的XCom Key：

from airflow.decorators import dag, task
from airflow.models.xcom_arg import XComArg
from datetime import datetime

@dag(start_date=datetime(2024,1,1), schedule=None, catchup=False)
def dynamic_map_dag():
    @task
    def get_name(ti):
        # 推送自定义XCom Key 'nicks'
        ti.xcom_push(key='nicks', value=["Jerry", "J-Rock", "J-Man"])
        ti.xcom_push(key='name', value="Jerald")

    @task
    def process_nick(nick):
        print(f"Processing nickname: {nick}")
        return f"Processed {nick}"

    # 用XComArg指定获取自定义key 'nicks'的XCom值
    nicks_list = XComArg(get_name(), key='nicks')
    process_nick.expand(nick=nicks_list)

dynamic_map_dag()

关键注意点 #

• 当任务return字典时，task_instance.output就是这个字典，直接通过键值访问（name_data["nicks"]）即可得到可迭代的列表，Airflow会自动识别用于动态映射。
• 使用自定义XCom Key时，必须显式用XComArg指定key，不能直接用get_name.output['nicks']，因为get_name.output默认指向return_value对应的XCom，而非自定义key的内容。

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