Spark from_json静默丢弃未定义字段，FAILFAST模式无效问题

问题背景 #

使用Spark从Kafka读取JSON数据，通过from_json()转换为Struct类型时指定了Schema，希望JSON中存在Schema未定义的字段时触发报错，但from_json()会静默丢弃这些字段。即使设置mode="FAILFAST"和columnNameOfCorruptRecord参数，也未触发报错或生成corrupt记录，执行结果依旧忽略了额外字段。

示例代码：

import pyspark.sql.functions as f
import pyspark.sql.types as t

data = [
    {
        "data": '{"key1": "value", "key2": "value"}',
    },
]

schema = t.StructType(
        [
            t.StructField("key1", t.StringType()),
            t.StructField("corrupt", t.StringType()),
        ]
)

df = spark.createDataFrame(data=data)

df = df.withColumn("data", f.from_json("data", schema, options={
    "mode": "FAILFAST",
    "columnNameOfCorruptRecord": "corrupt"
}))

df.printSchema()
df.show()

执行结果：

root
 |-- data: struct (nullable = true)
 |    |-- key1: string (nullable = true)
 |    |-- corrupt: string (nullable = true)

+-------------+
|         data|
+-------------+
|{value, NULL}|
+-------------+

原因分析 #

Spark的from_json()函数中，FAILFAST模式仅针对JSON格式错误、字段类型不匹配、必填字段缺失这类解析层面的错误。而JSON中存在Schema未定义的额外字段，默认被视为合法场景，不会触发FAILFAST逻辑，Spark会直接忽略这些字段。同时，columnNameOfCorruptRecord仅用于存储解析失败的原始JSON字符串，也不会捕捉到额外字段的情况。

解决方案 #

方法1：解析为Map后校验字段 #

先将JSON字符串解析为Map类型，提取所有键并与目标Schema的字段名对比，若存在额外字段则触发报错或标记为corrupt记录。

代码示例：

import pyspark.sql.functions as f
import pyspark.sql.types as t

data = [
    {
        "data": '{"key1": "value", "key2": "value"}',
    },
]

# 定义目标业务Schema
target_schema = t.StructType([t.StructField("key1", t.StringType())])
# 提取Schema字段名列表
schema_field_names = [field.name for field in target_schema.fields]

df = spark.createDataFrame(data=data)

# 1. 将JSON解析为Map，提取所有键
df = df.withColumn("json_map", f.from_json("data", t.MapType(t.StringType(), t.StringType())))
df = df.withColumn("json_keys", f.map_keys("json_map"))

# 2. 检查是否存在Schema未定义的额外字段
df = df.withColumn(
    "has_extra_fields",
    f.size(f.array_except(f.col("json_keys"), f.array(*[f.lit(name) for name in schema_field_names]))) > 0
)

# 3. 处理额外字段：可选两种方式
# 方式A：直接抛出异常终止任务
extra_fields_df = df.filter(f.col("has_extra_fields"))
if extra_fields_df.count() > 0:
    extra_keys = extra_fields_df.select(f.collect_set("json_keys")).first()[0]
    raise ValueError(f"检测到未定义的JSON字段: {extra_keys}")

# 方式B：标记corrupt记录并保留原始数据
df = df.withColumn(
    "parsed_data",
    f.when(f.col("has_extra_fields"), f.lit(None).cast(target_schema))
    .otherwise(f.from_json("data", target_schema))
)
df = df.withColumn(
    "corrupt_record",
    f.when(f.col("has_extra_fields"), f.col("data")).otherwise(f.lit(None))
)

df.select("parsed_data", "corrupt_record").show()

方法2：利用动态Schema对比（Spark 3.0+） #

通过schema_of_json函数获取JSON的动态Schema，与目标Schema对比，若存在额外字段则触发报错。

代码示例：

import pyspark.sql.functions as f
import pyspark.sql.types as t

data = [
    {
        "data": '{"key1": "value", "key2": "value"}',
    },
]

target_schema = t.StructType([t.StructField("key1", t.StringType())])
target_field_names = set(field.name for field in target_schema.fields)

df = spark.createDataFrame(data=data)

# 获取JSON的动态Schema
dynamic_schema_str = df.select(f.schema_of_json(f.col("data"))).first()[0]
dynamic_schema = t.StructType.fromJson(dynamic_schema_str)
dynamic_field_names = set(field.name for field in dynamic_schema.fields)

# 检查额外字段
extra_fields = dynamic_field_names - target_field_names
if extra_fields:
    raise ValueError(f"JSON包含未定义字段: {extra_fields}")

# 正常解析
df = df.withColumn("data", f.from_json("data", target_schema))
df.show()

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