PySpark/Python处理嵌套JSON补全缺失键空值解决字段不存在报错

报错核心是两个配置错误：

• 你使用的multiline=true配置仅适用于整个文件为单个JSON对象/JSON数组的场景，你当前的文件是每行一个独立JSON对象，开这个配置会导致Spark仅按解析到的第一个JSON块推断schema，后续行出现的key1、key3根本不会被纳入data struct的结构定义
• 默认JSON读取的采样比例不是100%，如果字段只出现在数据靠后的位置，也可能漏识别

之前尝试预定义schema、when判断没生效的原因：

• 开着multiline模式的解析逻辑和逐行JSON冲突，传入的schema无法正常匹配解析逻辑
• when判断的前提是目标字段已经存在于schema中，字段未被定义时，判断语句本身就会先抛出字段不存在的异常

1. 修正读取配置 #

逐行JSON必须关闭multiline选项，如需自动推断全量字段，把采样比例设为1.0即可：

df_landing = spark.read.format("json") \
    .option("multiline", "false") \
    .option("samplingRatio", 1.0) \
    .load(input_file)

执行后打印的schema会自动包含所有字段：

root 
|-- data: struct (nullable = true) 
| |-- key0: string (nullable = true)
| |-- key1: string (nullable = true)
| |-- key2: string (nullable = true)
| |-- key3: string (nullable = true) 
|-- id: long (nullable = true) 
|-- type: string (nullable = true)

2. 生产环境建议显式传入完整schema #

自动推断依赖数据扫描，生产环境提前定义固定schema更稳定，不会因为数据异常丢字段：

from pyspark.sql.types import StructType, StructField, StringType, LongType

target_schema = StructType([
    StructField("id", LongType(), True),
    StructField("type", StringType(), True),
    StructField("data", StructType([
        StructField("key0", StringType(), True),
        StructField("key1", StringType(), True),
        StructField("key2", StringType(), True),
        StructField("key3", StringType(), True)
    ]), True)
])

df_landing = spark.read.format("json") \
    .option("multiline", "false") \
    .schema(target_schema) \
    .load(input_file)

3. 扁平化提取字段 #

schema修正后直接选取字段即可，不会再抛出struct字段不存在的报错：

df_flatten = df_landing.select(
    "id",
    "type",
    "data.key0",
    "data.key1",
    "data.key2",
    "data.key3"
)

df_flatten.show()

执行输出结果如下，不存在的字段会自动填充null：

+---+----+----+----+----+----+
| id|type|key0|key1|key2|key3|
+---+----+----+----+----+----+
|  1| int|val1|null|val2|null|
|  2| int|null|null|val3|val4|
|  3| int|null|val5|null|val6|
+---+----+----+----+----+----+

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