Spark 2.7加载多行JSON：单条脏数据致全文件被标记为脏记录的问题

问题本质 #

启用multiline=true时，Spark会将整个JSON数组文件视为单条JSON记录。一旦数组内任意元素不符合指定Schema，整个数组的解析就会失败，导致全文件内容被存入_corrupt_record，而非仅标记单个脏元素。这和单行JSON（每个对象占一行）的解析逻辑不同——单行模式下每个对象是独立记录，脏数据只会影响自身。

解决方案 #

方案1：先读取为字符串，再拆分解析数组

通过手动解析JSON数组，实现对单个脏元素的捕获：

1. 读取整个文件为字符串记录

   // 不指定Schema，先读取文件内容为单条字符串记录
   Dataset<Row> rawDf = spark.read()
       .option("multiline", "true")
       .text("filepath");
   

2. 解析字符串为JSON数组，指定错误处理规则
   把原元素Schema包装为ArrayType，用from_json函数解析时配置错误处理参数：

   import org.apache.spark.sql.functions;
   import org.apache.spark.sql.types.DataTypes;
   import java.util.HashMap;
   
   // 构建数组类型的Schema（外层为ArrayType，内层为原自定义StructType）
   StructType arraySchema = DataTypes.createArrayType(schema);
   
   // 解析时启用PERMISSIVE模式，指定脏记录列名
   Dataset<Row> parsedDf = rawDf.withColumn("json_array", functions.from_json(
       functions.col("value"),
       arraySchema,
       new HashMap<String, String>() {{
           put("mode", "PERMISSIVE");
           put("columnNameOfCorruptRecord", "_corrupt_record");
       }}
   ));
   

3. 展开数组为多行记录
   将数组拆分为独立行，分离正常数据和脏记录：

   // 展开数组，得到每条对象的记录
   Dataset<Row> explodedDf = parsedDf.select(functions.explode(functions.col("json_array")).alias("data"))
       .select("data.*");
   
   // 若需保留脏记录列，可调整为：
   Dataset<Row> fullDf = parsedDf.select(
       functions.explode(functions.col("json_array")).alias("data"),
       functions.col("json_array._corrupt_record")
   ).select("data.*", "_corrupt_record");
   

方案2：预处理文件转为单行JSON格式

如果允许修改输入文件，可将JSON数组拆分为每行一个JSON对象（移除外层[]，每个对象末尾添加换行），之后直接用原单行JSON的解析逻辑即可，脏数据只会被单独标记。

补充说明 #

Spark 2.7版本中，multiline模式的JSON解析器对数组类型的错误处理存在局限性——它将整个文件视为单一JSON实体，而非多个独立对象的集合。上述方案通过手动拆分解析的方式，绕开了这个限制，实现了对数组内单个脏元素的精准捕获。

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