使用Apache Spark读取混合转义JSON的数组时报_corrupt_record错误如何解决

Spark内置JSON读取器的解析规则和当前文件的混合结构完全不兼容，三种尝试全部触发规则匹配失败：

• 默认读取模式要求逐行对应一个独立的裸JSON对象，不支持顶层JSON数组结构。你文件的第一行就是数组开头加转义字符串，完全不符合单JSON对象的语法，直接被判定为损坏记录。
• 用wholeTextFiles取全量文本再喂给JSON读取的方式，是把整个文件的内容当成单条记录解析，既不会自动展开顶层JSON数组为多行，又遇到混合了字符串化JSON元素的结构，无法推断列结构，自然触发损坏记录报错。
• 开启multiline选项的读取方式，虽然支持跨多行解析JSON结构，能识别出顶层是个数组，但解析数组元素时遇到第一个元素是被转义的字符串字面量（本质是字符串，不是可以提取字段的裸JSON对象），无法匹配预期的对象结构，最终把全量内容标记为_corrupt_record。

本质上，这个文件的JSON数组里混了两类完全不同的元素：一类是转成字符串的JSON对象（外层带双引号、内部双引号被转义符转义），一类是正常的裸JSON对象，这正是你判断的报错核心来源，没有任何内置读取参数能直接兼容这种混合结构。

整个过程不要上来就把全量内容喂给Spark的内置JSON解析器，先做文本拆分、类型适配，再做结构化解析，步骤如下：

1. 读取文件原始文本，拆分出独立的数组元素
   用wholeTextFiles读取文件全量文本，先去掉顶层包裹数组的[和]，再根据数组元素的分隔位置拆分出每个独立的内容块，避免把字符串内部的逗号误判为元素分隔符：

   val rawElementRDD = spark.sparkContext.wholeTextFiles("你的文件路径.json")
     .values
     .flatMap(fileContent => {
       val cleanedContent = fileContent.trim.stripPrefix("[").stripSuffix("]")
       // 仅在两个数组元素的分隔位置拆分，不切分JSON内部的逗号
       cleanedContent.split("(?<=\"\\}),\\s*(?=\\{)|(?<=\\}\\}),\\s*(?=\\{)")
         .map(_.trim.stripSuffix(",").trim)
     })
   

2. 定义适配混合结构的解析UDF
   针对两类元素的差异做适配：对带外层引号的转义JSON字符串，先剥掉外层引号、把转义的双引号还原为正常格式，再解析成标准JSON字符串；对本来就是裸JSON对象的元素，直接解析为标准JSON字符串：

   import org.apache.spark.sql.functions.udf
   import org.json4s.jackson.JsonMethods.parse
   // 注册解析UDF
   val normalizeMixedJson = udf((rawBlock: String) => {
     implicit val format = org.json4s.DefaultFormats
     rawBlock match {
       // 匹配被引号包裹的转义JSON字符串
       case block if block.startsWith("\"") && block.endsWith("\"") =>
         parse(
           block.stripPrefix("\"")
             .stripSuffix("\"")
             .replace("\\\"", "\"")
         ).toString
       // 匹配正常裸JSON对象
       case block => parse(block).toString
     }
   })
   

3. 生成可被Spark识别的结构化DataFrame
   先把拆分后的元素转成单列表格，应用UDF把所有内容统一成标准JSON字符串格式，最后把统一后的JSON内容喂给Spark内置JSON解析器自动推断列结构：

   // 把原始块转成DataFrame
   val rawDF = rawElementRDD.toDF("raw_block")
   // 应用UDF做格式归一
   val normalizedDF = rawDF.withColumn("json_content", normalizeMixedJson(col("raw_block")))
   // 用Spark内置解析器生成最终结构化DataFrame
   val finalStructuredDF = spark.read.json(normalizedDF.select("json_content").as[String])
   

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