Airflow中K8sPodOperator传递XCOM JSON双引号丢失致JSONDecodeError求助

问题根源 #

当你将XCOM中的JSON字符串作为命令行参数传递给KubernetesPodOperator时，K8s的shell会解析参数中的双引号，将原本的{"agents": ["john.doe@example.com"]}解析为{agents: [john.doe@example.com]}，丢失了JSON格式必需的双引号，最终导致json.loads解析失败。

解决方案 #

方案1：使用Jinja的tojson过滤器转义 #

修改XCOM数据的Jinja表达式，通过tojson过滤器将JSON字符串转义为适合命令行传递的格式：

XCOM_DATA = '{{ task_instance.xcom_pull(task_ids="task_name", key="return_value")["data"] | tojson }}'

tojson会自动将内部的双引号转义为\"，确保shell解析时不会破坏JSON结构。

方案2：通过环境变量传递JSON（推荐） #

避免命令行参数的shell解析问题，直接将JSON字符串注入到Pod的环境变量中：

KubernetesPodOperator(
    cmds=["python", "src/send_notification.py"],
    env_vars={
        "DATA_JSON": '{{ task_instance.xcom_pull(task_ids="task_name", key="return_value")["data"] }}'
    },
    task_id="notify",
    name="notify",
    dag=dag,
    **COMMON_TASKS_ARGS,
)

在脚本中通过环境变量读取并解析：

import os
import json

# 从环境变量读取JSON字符串并解析
data = json.loads(os.environ.get("DATA_JSON", "{}"))

方案3：手动转义双引号 #

使用Jinja的replace过滤器手动将双引号替换为转义后的\"：

XCOM_DATA = '{{ task_instance.xcom_pull(task_ids="task_name", key="return_value")["data"] | replace('"', '\\"') }}'

方案4：使用Jinja的quote过滤器 #

quote过滤器会将字符串处理为符合shell参数规范的格式，自动处理引号转义：

XCOM_DATA = '{{ task_instance.xcom_pull(task_ids="task_name", key="return_value")["data"] | quote }}'

验证 #

修改后，检查传递给K8s的命令，确保JSON字符串的双引号被正确保留（例如显示为"{\"agents\": [\"john.doe@example.com\"]}"），此时脚本中的json.loads即可正常解析。

内容的提问来源于stack exchange，提问作者Alexander Montoya Ángel