如何在Airflow中使用XComArgs语法获取嵌套XCom输出？

我之前也遇到过一模一样的问题，核心是对XComArgs.output的用法有个小误解，再加上配置细节没踩对，咱们一步步来解决：

关键错误点：多了不必要的['return_value'] #

你写的task_1.output['return_value']['data']['foo'][0]['cmd']里，['return_value']是完全多余的！

因为task_1.output本身就等价于ti.xcom_pull(task_ids='task_1', key='return_value')的结果——它直接指向了task_1返回值对应的XCom数据。你额外加的['return_value']相当于在返回的字典{"data": {"foo": [{"cmd": "ls"}]}}里找return_value这个键，显然不存在，所以才返回null。

正确的实现步骤 #

1. 确保DAG开启原生对象渲染 #

首先必须在DAG配置里设置render_template_as_native_obj=True，这个参数会让Airflow把XCom的JSON数据反序列化为原生Python字典/列表，而不是字符串，这样才能直接进行嵌套下标访问。

2. 正确使用XComArgs访问嵌套结构 #

下面是完整的示例代码，用传统PythonOperator演示：

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.utils.dates import days_ago

# 上游任务返回嵌套字典
def get_nested_data():
    return {"data": {"foo": [{"cmd": "ls"}]}}

with DAG(
    dag_id="nested_xcom_demo",
    start_date=days_ago(1),
    render_template_as_native_obj=True,  # 必须开启这个配置
    catchup=False
) as dag:
    task_1 = PythonOperator(
        task_id="task_1",
        python_callable=get_nested_data
    )

    # 下游任务接收嵌套XCom值
    def process_cmd(cmd):
        print(f"Executing command: {cmd}")
        return f"Successfully ran: {cmd}"

    task_2 = PythonOperator(
        task_id="task_2",
        python_callable=process_cmd,
        op_kwargs={
            # 正确写法：直接用task_1.output访问嵌套结构
            "cmd": task_1.output["data"]["foo"][0]["cmd"]
        }
    )

    task_1 >> task_2

3. 如果是自定义XCom键的情况 #

如果你的上游任务不是用return_value，而是手动push了自定义键的XCom（比如ti.xcom_push(key="custom_key", value=...)），那需要用XComArg类指定键来获取：

from airflow.models.xcom_arg import XComArg

def push_custom_xcom(ti):
    ti.xcom_push(key="custom_data", value={"data": {"foo": [{"cmd": "ls"}]}})

task_1 = PythonOperator(
    task_id="task_1",
    python_callable=push_custom_xcom
)

# 获取自定义键的XCom
custom_xcom = XComArg(task_1, key="custom_data")

task_2 = PythonOperator(
    task_id="task_2",
    python_callable=process_cmd,
    op_kwargs={
        "cmd": custom_xcom["data"]["foo"][0]["cmd"]
    }
)

验证方法 #

你可以在Airflow UI里查看XCom数据：进入task_1的实例详情，切换到XCom标签，确认return_value对应的是你返回的嵌套字典，而不是字符串。如果是字符串，说明render_template_as_native_obj=True没生效，检查DAG配置是否正确。

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