如何在Spark中检查JSON字符串是否包含Schema未定义的额外字段

当使用Spark SQL的from_json函数时，若原始JSON包含Schema未定义的字段，这些字段会被直接忽略。以下是几种检测这类额外字段的实用方法：

方法一：利用Map类型提取所有键并对比Schema字段 #

将原始JSON字符串转换为MapType类型（会保留所有键值对），再提取所有键与Schema定义的字段做对比：

import org.apache.spark.sql.functions._
import org.apache.spark.sql.types._

// 定义目标Schema
val targetSchema = StructType(Seq(StructField("field0", StringType)))

// 构造示例DataFrame
val rawDF = spark.createDataFrame(Seq(("{\"field0\":\"go\",\"field1\":\"test\"}",))).toDF("json_string")

// 提取JSON中所有字段名
val dfWithAllKeys = rawDF.withColumn(
  "all_json_keys",
  map_keys(from_json(col("json_string"), MapType(StringType, StringType)))
)

// 获取Schema定义的字段集合
val schemaFieldSet = targetSchema.fieldNames.toSet

// 计算存在的额外字段
val resultDF = dfWithAllKeys.withColumn(
  "extra_fields",
  array_except(col("all_json_keys"), lit(schemaFieldSet.toArray))
)

resultDF.show(false)

执行后会输出每个JSON字符串对应的额外字段列表，空数组表示无额外字段。

方法二：动态推断JSON Schema并与自定义Schema对比 #

通过schema_of_json函数推断原始JSON的完整Schema，再与自定义Schema对比找出差异字段：

import org.apache.spark.sql.functions._
import org.apache.spark.sql.types._

// 构造示例DataFrame
val rawDF = spark.createDataFrame(Seq(("{\"field0\":\"go\",\"field1\":\"test\"}",))).toDF("json_string")

// 从样本JSON推断完整Schema
val inferredSchemaStr = schema_of_json(rawDF.select("json_string").head().getString(0))
val inferredSchema = DataType.fromJson(inferredSchemaStr).asInstanceOf[StructType]

// 自定义目标Schema
val targetSchema = StructType(Seq(StructField("field0", StringType)))

// 找出额外字段
val extraFields = inferredSchema.fieldNames.filterNot(targetSchema.fieldNames.contains(_))

println(s"检测到的额外字段: ${extraFields.mkString(", ")}")

注意：若数据集包含多种JSON结构，建议取包含所有可能字段的样本进行推断，避免遗漏。

方法三：自定义UDF解析JSON（适用于复杂嵌套结构） #

对于嵌套层级较深的JSON，可通过自定义UDF结合JSON解析库（如Jackson）提取所有字段路径，再与Schema的字段路径对比：

import com.fasterxml.jackson.databind.JsonNode
import com.fasterxml.jackson.databind.ObjectMapper
import org.apache.spark.sql.api.java.UDF1
import org.apache.spark.sql.functions._
import scala.collection.JavaConverters._

// 注册UDF：提取JSON所有字段路径（支持嵌套）
spark.udf.register("extract_all_fields", new UDF1[String, Array[String]] {
  private val mapper = new ObjectMapper()
  private def extractFields(node: JsonNode, path: String = ""): List[String] = {
    if (node.isObject) {
      node.fieldNames().asScala.flatMap { key =>
        val currentPath = if (path.isEmpty) key else s"$path.$key"
        extractFields(node.get(key), currentPath)
      }.toList
    } else if (node.isArray) {
      List(path) // 数组类型直接记录路径
    } else {
      List(path)
    }
  }

  override def call(jsonStr: String): Array[String] = {
    val jsonNode = mapper.readTree(jsonStr)
    extractFields(jsonNode).toArray
  }
}, ArrayType(StringType))

// 定义目标Schema的字段路径（嵌套字段用点分隔，如"parent.child"）
val targetFieldPaths = Array("field0")

// 检测额外字段
val resultDF = rawDF.select(
  col("json_string"),
  array_except(callUDF("extract_all_fields", col("json_string")), lit(targetFieldPaths)).alias("extra_fields")
)

resultDF.show(false)

该方法能处理嵌套JSON结构，精准识别所有未在Schema中定义的字段路径。

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