Databricks中PySpark DataFrame提取嵌套JSON内容返回Null的解决咨询

你的提取结果返回Null，核心原因是**json_col中的内容并非标准JSON格式**，存在多处语法错误：

• 键名未用双引号包裹（比如html:null应为"html":null）
• 存在多余连续逗号（比如search_alias:{...},,content中的双逗号）
• 数组元素语法错误（all_images最后一个元素缺少外层大括号）
• 末尾存在多余逗号（比如climate_pledge_friendly":"No",}中的逗号）

这些错误导致get_json_object无法正常解析，直接返回Null。

方案一：修复JSON格式后提取 #

先通过正则修正非标准JSON的语法问题，再进行内容提取：

1. 修复JSON格式 #

from pyspark.sql.functions import col, regexp_replace, get_json_object

# 给所有键名添加双引号
fixed_json = product.withColumn(
    "fixed_json_col",
    regexp_replace(col("json_col"), r"(\w+):", r'"\1":')
)

# 移除连续多余的逗号
fixed_json = fixed_json.withColumn(
    "fixed_json_col",
    regexp_replace(col("fixed_json_col"), r",,", r",")
)

# 修复all_images数组最后一个元素的语法错误（补全大括号）
fixed_json = fixed_json.withColumn(
    "fixed_json_col",
    regexp_replace(col("fixed_json_col"), r',"dasdas":"dkasjhasdasnd","name":"diasdjaskldnasn"(\])', r',{"dasdas":"dkasjhasdasnd","name":"diasdjaskldnasn"}\1')
)

# 移除末尾多余的逗号
fixed_json = fixed_json.withColumn(
    "fixed_json_col",
    regexp_replace(col("fixed_json_col"), r",(\})", r"\1")
)

2. 提取目标content内容 #

extracted_data = fixed_json.withColumn(
    "content_extract",
    get_json_object(col("fixed_json_col"), "$.product.content")
)

# 查看结果（关闭截断以显示完整内容）
extracted_data.select("content_extract").show(truncate=False)

方案二：用Schema解析（更灵活） #

如果需要后续对提取的内容做结构化处理，推荐使用from_json配合Schema解析：

1. 定义Schema #

from pyspark.sql.types import StructType, StructField, StringType, ArrayType, MapType

# 定义content的结构Schema
content_schema = StructType([
    StructField("all_images", ArrayType(MapType(StringType(), StringType()))),
    StructField("body_text", StringType())
])

# 定义整个JSON的完整Schema
full_schema = StructType([
    StructField("product", StructType([
        StructField("content", content_schema)
    ]))
])

2. 解析并提取 #

from pyspark.sql.functions import from_json

# 先执行方案一中的JSON格式修复步骤，得到fixed_json DataFrame
# ...（此处省略重复的修复代码）

# 解析标准JSON为结构化数据
parsed_df = fixed_json.withColumn(
    "parsed_json",
    from_json(col("fixed_json_col"), full_schema)
)

# 提取目标content部分
extracted_data = parsed_df.withColumn(
    "content_extract",
    col("parsed_json.product.content")
)

extracted_data.select("content_extract").show(truncate=False)

如果原始数据中存在更多类似的格式问题，需要调整正则表达式覆盖对应场景；若数据量极大，建议先在源头规范JSON生成逻辑，避免后续修复的性能损耗。

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