如何在Airflow中反序列化XCom字符串以实现任务间数据交互？

假设有一个包含两个任务的DAG：DAG: Task A >> Task B（使用BashOperator或DockerOperator），二者需要通过XCom完成通信：

• Task A通过标准输出输出单行JSON格式的信息，该内容可在任务日志中查看。当设置xcom_push=True时，输出内容会被存入Task A的return_value XCom键中，示例输出内容：{"key1":1,"key2":3}
• Task B仅需获取Task A输出的key2值，尝试通过Jinja模板{{xcom_pull('task_a')['key2']}}直接传递，但触发报错：jinja2.exceptions.UndefinedError: 'str object' has no attribute 'key2'——原因是XCom存储的return_value只是原始JSON字符串，而非可直接取值的字典对象。

希望能实现类似Airflow Variables的用法：在Jinja模板中直接对XCom的JSON内容进行反序列化并提取字段（比如{{ var.json.my_var.path }}这种便捷写法）。

临时解决方案 #

目前可以通过在Task A中将JSON字符串转换为Python字典后再推送至XCom来解决问题，示例实现如下：

用PythonOperator封装JSON处理（替代/配合Bash/Docker输出） #

如果原Task A是Bash或DockerOperator，可以新增一个Python任务来处理输出并推送结构化XCom，或者直接用PythonOperator完成逻辑：

from airflow.operators.python import PythonOperator
import json
from airflow import DAG
from datetime import datetime

def parse_and_push_json(**context):
    # 模拟Bash/DockerOperator的输出结果
    bash_output = '{"key1":1,"key2":3}'
    # 反序列化为字典
    output_dict = json.loads(bash_output)
    # 推送结构化数据到XCom
    context["ti"].xcom_push(key="return_value", value=output_dict)

with DAG(
    dag_id="json_xcom_example",
    start_date=datetime(2024, 1, 1),
    schedule_interval=None
) as dag:
    task_a = PythonOperator(
        task_id="task_a",
        python_callable=parse_and_push_json,
        provide_context=True
    )

    # Task B 示例（以BashOperator为例）
    task_b = BashOperator(
        task_id="task_b",
        bash_command='echo "Key2 value is: {{ xcom_pull(\'task_a\')[\'key2\'] }}"'
    )

    task_a >> task_b

这样Task B的Jinja模板就能直接提取key2的值，不会再触发字符串属性错误。

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