Scala解析JSON异常处理：Spark DataFrame提取JSON块遇类型错误求方案

你猜的没错！这个报错确实是因为raw_json列里的数据不统一——有的是符合你定义Schema的JSON，有的却是普通字符串、空值或者格式无效的JSON，导致from_json直接罢工了。下面给你几个实用的解决方案，分场景来选：

方案1：用Spark 3.0+的try函数优雅容错 #

如果你的Spark版本是3.0及以上，直接用内置的try函数包裹from_json，解析失败时自动返回null，不用自己写复杂的异常处理：

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

// 先定义你需要的目标Schema
val jsonSchema = StructType(Seq(
  StructField("json_blob", StructType(Seq(
    StructField("my_field", MapType(StringType, StringType, true))
  )))
))

// 使用try函数包裹解析逻辑，失败则返回null
val mydf = mydf_withjson
  .withColumn("json", try(from_json($"raw_json", jsonSchema), lit(null).cast(jsonSchema)))
  .select("*", "json.*")

方案2：自定义UDF兼容旧版Spark（3.0以下） #

要是你还在用Spark 3.0之前的版本，就写个简单的UDF来捕获解析异常，同样让失败的行返回null：

import org.apache.spark.sql.functions._
import org.apache.spark.sql.types._
import scala.util.Try

// 自定义安全解析JSON的UDF
val safeFromJson = udf((jsonStr: String) => {
  Try(from_json(lit(jsonStr), jsonSchema).asInstanceOf[Row]).toOption.orNull
})

// 应用UDF完成解析
val mydf = mydf_withjson
  .withColumn("json", safeFromJson($"raw_json"))
  .select("*", "json.*")

方案3：提前过滤无效数据 #

如果不想保留解析失败的行，可以先过滤掉空值和明显不是JSON的内容（比如简单判断是否以{开头、}结尾）：

// 过滤空值和非JSON格式的行
val filteredDf = mydf_withjson
  .filter($"raw_json".isNotNull && $"raw_json".rlike("^\\{.*\\}$"))

// 再进行正常解析
val mydf = filteredDf
  .withColumn("json", from_json($"raw_json", jsonSchema))
  .select("*", "json.*")

方案4：兼容json_blob为字符串的场景 #

如果你的json_blob字段本身就可能是字符串或者嵌套结构（而非单纯解析失败），那可以用Spark 3.1+支持的UnionType来定义Schema，直接兼容两种类型：

// 定义支持字符串和嵌套结构的Schema
val jsonSchema = StructType(Seq(
  StructField("json_blob", UnionType(
    StringType,
    StructType(Seq(StructField("my_field", MapType(StringType, StringType, true))))
  ))
))

// 解析后根据类型处理数据
val mydf = mydf_withjson
  .withColumn("json", try(from_json($"raw_json", jsonSchema), lit(null).cast(jsonSchema)))
  // 区分json_blob是字符串还是结构，按需处理
  .withColumn("my_field", when(
    $"json.json_blob".cast(StringType).isNotNull,
    // 这里可以根据需求处理字符串场景，比如转成Map或保留原字符串
    lit(null) // 示例：如果是字符串就返回null，你可以改成自己的业务逻辑
  ).otherwise(
    $"json.json_blob.my_field"
  ))
  .select("*", "my_field")

总结一下：核心就是给解析逻辑加容错，要么用内置函数，要么自定义处理；如果字段本身有多种类型，就用UnionType来兼容，这样就能解决数据不统一的问题啦！

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