如何在PySpark中解析带动态Schema的宽松格式JSON

核心问题拆解 #

• PySpark原生from_json仅支持严格JSON格式（键必须用双引号包裹），直接解析无引号键的宽松JSON会导致所有嵌套列返回NULL
• 动态变化的键（如question5、question7）及对应值的结构无法通过固定Schema定义，需采用动态结构处理方案

解决方案步骤 #

1. 宽松JSON转严格JSON #

PySpark无内置宽松JSON解析器，通过Python UDF调用第三方库（demjson或json5）完成格式转换：

• 先在所有Spark节点安装依赖：pip install demjson
• 编写UDF将宽松JSON字符串转为严格JSON，或直接解析为Python字典（PySpark自动映射为MapType）

2. 动态键值对处理 #

由于键和值的结构均动态变化，优先将JSON转为MapType(StringType, StringType)，再根据值的特征进一步解析；也可将键值对拆分为多行，方便后续按需处理。

完整示例代码 #

假设数据集包含列loose_json_col，存储内容类似：{question5: {answer: "yes", score: 5}, question7: {option: "A", comment: "none"}}

from pyspark.sql import SparkSession
from pyspark.sql.functions import udf, explode, from_json
from pyspark.sql.types import MapType, StringType, StructType, StructField
import demjson

# 初始化SparkSession
spark = SparkSession.builder.appName("LooseJsonParser").getOrCreate()

# 构造示例数据
sample_data = [
    ("{question5: {answer: \"yes\", score: 5}, question7: {option: \"A\", comment: \"none\"}}",),
    ("{question3: {value: 100}, question9: {text: \"test\", valid: true}}",)
]
df = spark.createDataFrame(sample_data, ["loose_json_col"])

# UDF：将宽松JSON转为严格JSON字符串
@udf(StringType())
def loose_to_strict(loose_str):
    try:
        parsed_dict = demjson.decode(loose_str)
        return demjson.encode(parsed_dict)
    except:
        return None

# 第一步：转换为严格JSON格式
df_strict = df.withColumn("strict_json", loose_to_strict("loose_json_col"))

# 第二步：解析为MapType处理动态键
map_schema = MapType(StringType(), StringType())
df_map = df_strict.withColumn("dynamic_kv", from_json("strict_json", map_schema))

# 可选：将动态键值对拆分为多行，便于后续处理
df_exploded = df_map.select("loose_json_col", explode("dynamic_kv").alias("question_key", "value_str"))

# 按需解析值的嵌套结构（示例：处理两种不同的value格式）
answer_schema = StructType([
    StructField("answer", StringType()),
    StructField("score", StringType())
])
option_schema = StructType([
    StructField("option", StringType()),
    StructField("comment", StringType())
])

@udf(answer_schema)
def parse_answer(value_str):
    try:
        return demjson.decode(value_str)
    except:
        return None

@udf(option_schema)
def parse_option(value_str):
    try:
        return demjson.decode(value_str)
    except:
        return None

df_final = df_exploded.withColumn("answer_detail", parse_answer("value_str")) \
                      .withColumn("option_detail", parse_option("value_str"))

# 查看结果
df_final.show(truncate=False)

简化方案：直接解析为字典 #

若无需保留严格JSON中间步骤，可直接用UDF将宽松JSON解析为Python字典，PySpark自动转为MapType：

@udf(MapType(StringType(), StringType()))
def direct_parse_loose(loose_str):
    try:
        return demjson.decode(loose_str)
    except:
        return None

df_simple = df.withColumn("dynamic_kv", direct_parse_loose("loose_json_col"))
df_simple.show(truncate=False)

性能优化提示 #

对于键仅包含字母/数字的简单场景，可跳过第三方库，用正则表达式给无引号键添加双引号，避免UDF性能损耗：

from pyspark.sql.functions import regexp_replace

df_regex = df.withColumn("strict_json", 
    regexp_replace(regexp_replace("loose_json_col", r"(\w+):", r'"\1":'), r"'", r'"'))

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