PySpark：在DataFrame列中设置JSON字符串的最佳方法

你可以基于固定模板复用JSON固定结构，逐行读取DataFrame其他列的值填充动态字段，最终序列化为字符串存入新列，仅需pandas和Python标准json库即可完成，不需要额外依赖。

1. 预定义固定结构模板 #

把示例中不需要动态修改的name、headers部分提前定义为模板，仅留values字段作为动态填充位，避免每行重复构造固定结构产生冗余计算：

import json
import pandas as pd

# 固定JSON结构模板
JSON_TEMPLATE = {
  "name": "",
  "headers": [
    {"name": "A", "dataType": "number"},
    {"name": "B", "dataType": "string"},
    {"name": "C", "dataType": "string"}
  ],
  "values": []
}

2. 编写行级JSON构造函数 #

函数接收DataFrame行对象作为入参，按照headers的字段顺序，读取行内对应列的取值填充到values数组中，最终序列化为标准JSON字符串返回：

注意：填充values时的取值顺序必须和headers里的字段顺序严格对应，否则会出现数据和声明类型不匹配的问题。

def generate_json_str(row):
    # 复制模板避免全局模板被修改
    current_obj = JSON_TEMPLATE.copy()
    # 替换下方列名为你DataFrame中实际对应的列名，顺序严格匹配A/B/C
    current_obj["values"] = [
        [row["numeric_col_for_A"], row["str_col_for_B"], row["str_col_for_C"]]
    ]
    # 序列化为JSON字符串，关闭ascii转义避免中文乱码
    return json.dumps(current_obj, ensure_ascii=False)

3. 批量生成目标列 #

调用DataFrame的行级apply方法，把生成的JSON字符串赋值给新列即可：

# 假设你的原始DataFrame变量名为df
df["json_storage_col"] = df.apply(generate_json_str, axis=1)

常见问题说明 #

• 如果行数据存在空值，提前用fillna做类型兼容的填充，避免序列化后的值类型和headers中声明的dataType冲突
• 如果单条JSON需要存储多组values，直接在current_obj["values"]的列表中追加对应长度的取值列表即可，结构和示例保持一致
• 如果不需要存储中文等非ASCII字符，可以去掉json.dumps里的ensure_ascii=False参数减少字符串体积

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