Pyspark中能否动态推断JSON字符串列的Struct schema并自动展平？

方案1：通用全量推断方案（兼容所有Spark版本） #

该方案通过扫描全量JSON数据自动识别所有存在的键和对应类型，无需手动定义Schema：

from pyspark.sql.functions import col, from_json

# 步骤1：提取JSON字符串列转为RDD，用于推断Schema
json_rdd = sparkDF.select("stringDictionary").rdd.map(lambda x: x[0])
# 步骤2：调用Spark JSON数据源能力自动推断全局Schema
inferred_schema = spark.read.json(json_rdd).schema

# 步骤3：用推断出的Schema解析JSON列并展开
df_parsed = sparkDF.withColumn("json_data", from_json(col("stringDictionary"), inferred_schema)) \
                   .select("*", "json_data.*") \
                   .drop("json_data") # 可按需删除中间列

display(df_parsed)

方案2：Databricks优化方案（适合大数据量场景） #

如果数据量过大，全量扫描推断Schema成本较高，可以先提取所有JSON键再生成Schema，性能更好：

from pyspark.sql.functions import col, from_json, explode, map_keys, map_type, StringType
from pyspark.sql.types import StructType, StructField

# 步骤1：将JSON临时解析为Map类型，提取所有去重后的键
all_keys = sparkDF.select(
    explode(map_keys(from_json(col("stringDictionary"), map_type(StringType(), StringType())))).alias("json_key")
).distinct().collect()

# 步骤2：根据所有键生成StructType Schema
inferred_schema = StructType([StructField(row.json_key, StringType(), True) for row in all_keys])

# 步骤3：解析并展开JSON列
df_parsed = sparkDF.withColumn("json_data", from_json(col("stringDictionary"), inferred_schema)) \
                   .select("*", "json_data.*") \
                   .drop("json_data")

display(df_parsed)

注意事项 #

• 方案1支持自动识别JSON中的非字符串类型（数值、布尔、嵌套结构等），无需额外调整
• 若JSON结构更新频率高，可定期执行Schema推断逻辑存储生成的Schema，日常任务直接复用即可，无需每次扫描全量数据
• 所有展开后的列默认允许为空，匹配不同行JSON键缺失的场景

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