PySpark嵌套JSON类型不匹配致Schema验证中断的解决方法咨询

使用PySpark读取JSON文件并按预定义Schema验证时遇到以下问题：

预定义Schema：

testschema = StructType([
    StructField("_corrupt_record", StringType(), True),
    StructField("a", StringType(), True),
    StructField("b", StringType(), True),
    StructField("c", StringType(), True),
    StructField("d", StringType(), True),
    StructField("e", StructType([
        StructField("e_1", IntegerType(), True)
    ]), True)
])

待读取的JSON内容：

{
    "e": {
        "e_1": "2"
    },
    "a": "a",
    "b": "b",
    "c": "c",
    "d": "d"
}

读取验证代码：

peopleDF = spark.read.json(json_input, schema=testschema, multiLine=True)

peopleDF.createOrReplaceTempView("people")

teenagerNamesDF = spark.sql("SELECT * FROM people")
display(teenagerNamesDF)

现象：由于e_1定义为IntegerType但JSON中是字符串类型，预期仅e_1为null，实际所有字段均为null；若将嵌套的e块移至JSON末尾，则e之前的字段能正常填充，仅e_1为null。推测嵌套JSON的类型不匹配会中断Schema验证，寻求解决办法。

原因说明 #

PySpark默认JSON解析器在遇到嵌套字段类型不匹配时，若该嵌套字段位于JSON前部，解析器会终止对当前行的后续字段解析，导致全字段为null；若嵌套字段在尾部，前部字段已完成解析则不受影响。

方案1：使用from_json手动解析（推荐） #

先将JSON读取为字符串，再通过from_json函数配合PERMISSIVE模式解析，确保仅不匹配字段为null，其余字段正常保留：

from pyspark.sql.functions import from_json

# 读取原始JSON为整串文本
raw_df = spark.read.text(json_input, wholetext=True)

# 指定Schema和解析模式，提取所有字段
peopleDF = raw_df.select(
    from_json("value", testschema, options={"mode": "PERMISSIVE"}).alias("data")
).select("data.*")

# 后续查询操作不变
peopleDF.createOrReplaceTempView("people")
teenagerNamesDF = spark.sql("SELECT * FROM people")
display(teenagerNamesDF)

方案2：先读取为字符串类型，再转换为目标类型 #

修改Schema将e_1先定义为StringType，读取后再转换为IntegerType，转换失败时自动置为null：

from pyspark.sql.functions import col, cast
from pyspark.sql.types import StructType, StructField, StringType, IntegerType

# 修改Schema，将e_1设为StringType
testschema = StructType([
    StructField("_corrupt_record", StringType(), True),
    StructField("a", StringType(), True),
    StructField("b", StringType(), True),
    StructField("c", StringType(), True),
    StructField("d", StringType(), True),
    StructField("e", StructType([
        StructField("e_1", StringType(), True)
    ]), True)
])

# 读取JSON
peopleDF = spark.read.json(json_input, schema=testschema, multiLine=True)

# 将e_1转换为IntegerType，转换失败则为null
peopleDF = peopleDF.withColumn("e.e_1", col("e.e_1").cast(IntegerType()))

# 后续查询操作不变
peopleDF.createOrReplaceTempView("people")
teenagerNamesDF = spark.sql("SELECT * FROM people")
display(teenagerNamesDF)

方案3：明确设置读取模式为PERMISSIVE #

虽然默认模式为PERMISSIVE，但显式指定可确保解析器尽可能保留有效字段，仅将错误内容写入_corrupt_record：

peopleDF = spark.read.json(
    json_input,
    schema=testschema,
    multiLine=True,
    mode="PERMISSIVE"
)

# 后续查询操作不变
peopleDF.createOrReplaceTempView("people")
teenagerNamesDF = spark.sql("SELECT * FROM people")
display(teenagerNamesDF)

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