PySpark如何将字符串列与转义JSON列合并为单个标准JSON列

问题说明 #

• 输入DataFrame字段说明：
  • 普通string类型列，示例值为Contact
  • escaped_json转义JSON列：Schema不固定，存储带反斜杠转义的JSON内容，格式示例为{\"id\":\"27\",\"person\":{\"firstName\":\"Dan\",\"lastName\":\"Jones\"}}
• 目标输出：仅包含string_with_regular_json字段的DataFrame，字段值为合并原普通列键值对、escaped_json解析后所有键值对的标准JSON，示例为{"string":"Contact","id":"27","person":{"firstName":"Dan","lastName":"Jones"}}
• 现有实现缺陷：当前仅能通过如下代码单独解析escaped_json为独立DataFrame，丢失了普通列与JSON内容的行对应关系，无法直接合并得到结果：

json_column_df = spark.read.json(input_df.rdd.map(lambda row: row.escaped_json))

实现方案 #

直接逐行处理合并即可，不需要单独解析JSON列后再做join，避免行错位和额外性能开销，兼容任意不固定Schema的转义JSON：

import json

def process_row(row):
    # 还原转义JSON为标准JSON格式
    # 若JSON值本身包含合法反斜杠，将replace("\\", "")替换为replace(r'\"', '"')，仅还原转义引号
    standard_json_str = row.escaped_json.replace("\\", "")
    json_obj = json.loads(standard_json_str)
    # 注入普通列键值对，有其他普通列需要合并时，直接在这里追加键值对即可
    json_obj["string"] = row.string
    # 转回标准JSON字符串返回
    return (json.dumps(json_obj),)

# 生成最终结果DataFrame
result_df = input_df.rdd.map(process_row).toDF(["string_with_regular_json"])

方案说明 #

• 逐行处理天然保证普通列和JSON内容的对应关系，不会出现数据错位
• 无需提前定义JSON Schema，适配任意结构的嵌套JSON输入
• 新增普通列时仅需要在process_row函数内追加对应键值对赋值逻辑即可，扩展成本极低
• 避免了单独解析JSON列后通过行ID关联的shuffle开销，性能更优

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