Spark：如何高效解析单行存储的亿级非标准JSON文件

针对你这种单行包含亿级独立JSON对象的文件，Spark默认的JSON reader无法直接解析，核心原因是它期望的是**JSON Lines格式（每行一个JSON对象）**或标准JSON数组。下面给出几个从高效到灵活的方案，适配不同场景：

方案1：字符串预处理+内置函数解析（推荐，性能最优） #

利用Spark的内置字符串函数，先将单行的多个JSON对象拆分成每行一个，再用标准JSON解析器处理，全程避免UDF，性能拉满：

步骤分解： #

1. 读取原始文件：因为整个文件是单行，用text格式读取，得到仅含value列的DataFrame（只有1行数据）：

   val rawDF = spark.read.text("path/to/your/large.json")
   // Python版本
   # rawDF = spark.read.text("path/to/your/large.json")
   

2. 拆分JSON对象：将单行字符串中合法的对象分隔符}{替换为}\n{（把多个对象拆成换行分隔的行），再按换行分割成数组，最后explode成多行单个JSON字符串：
   这里的关键是用正则匹配不在字符串内部的}{（避免误分割字符串里的内容），正则表达式可以用：(?<=(?<!\\)(\\\\)*}){(?!")（匹配未被转义引号包裹的}{）

   import org.apache.spark.sql.functions.{regexp_replace, split, explode}
   
   val splitDF = rawDF
     .select(regexp_replace($"value", "(?<=(?<!\\\\)(\\\\\\\\)*}){(?!\")", "}\\n{").alias("split_str"))
     .select(split($"split_str", "\\n").alias("json_arr"))
     .select(explode($"json_arr").alias("json_str"))
   // Python版本
   # from pyspark.sql.functions import regexp_replace, split, explode
   
   # splitDF = rawDF \
   #   .select(regexp_replace("value", r'(?<=(?<!\\)(\\\\)*}){(?!")', "}\\n{").alias("split_str")) \
   #   .select(split("split_str", "\\n").alias("json_arr")) \
   #   .select(explode("json_arr").alias("json_str"))
   

3. 解析JSON字符串：用from_json配合你定义的Schema来解析每个JSON对象（提前定义Schema能大幅提升性能，避免Spark自动推断Schema的开销）：

   import org.apache.spark.sql.types.{StructType, StructField, StringType}
   
   // 假设你的JSON结构的Schema，实际根据你的嵌套结构定义
   val jsonSchema = new StructType()
     .add(StructField("id", StringType))
     .add(StructField("key", StringType))
     // 添加其他字段和嵌套结构
   
   val finalDF = splitDF.select(from_json($"json_str", jsonSchema).alias("data")).select("data.*")
   // Python版本
   # from pyspark.sql.types import StructType, StructField, StringType
   
   # jsonSchema = StructType([
   #   StructField("id", StringType()),
   #   StructField("key", StringType())
   #   # 添加其他字段和嵌套结构
   # ])
   
   # finalDF = splitDF.select(from_json("json_str", jsonSchema).alias("data")).select("data.*")
   

性能优势： #

• 全程使用Spark内置函数，避免UDF的序列化/反序列化开销
• 利用Spark的分布式计算能力，拆分后的数据可以并行解析
• 提前定义Schema避免了Spark全量扫描推断Schema的耗时

方案2：自定义InputFormat（极致性能，适合超大规模数据） #

如果你的文件大到方案1的字符串处理仍有瓶颈，可以自定义一个InputFormat，直接在读取阶段就逐个解析JSON对象，跳过字符串拆分的步骤：

核心思路： #

• 继承Spark的TextInputFormat，重写nextKeyValue方法，逐个读取字节流中的JSON对象（通过匹配{和}的层级来确定单个对象的边界）
• 这样读取后直接得到每行一个JSON对象的RDD，再转成DataFrame解析

注意事项： #

• 实现起来需要一定的Scala/Java基础，但性能是最优的，因为避免了大字符串的处理和拆分
• 要处理好JSON对象的嵌套层级（比如对象内部的{}不能当作边界）

方案3：预处理文件（离线场景适用） #

如果你的文件是静态的，不需要实时处理，可以先在离线环境用高效的命令行工具预处理，比如用jq或者自定义脚本将单行多对象转成JSON Lines格式：

# 用jq处理（需要确保jq支持大文件）
jq -c '.' your_large.json > formatted.json

不过1亿条数据的话，jq可能会比较慢，不如用Spark的方案高效，但适合一次性处理的场景。

避坑提醒： #

• 不要尝试将整个文件读入内存（比如用collect获取单行字符串），1亿条数据的字符串会直接撑爆内存
• 必须提前定义Schema，自动推断Schema需要扫描全量数据，耗时极长且容易OOM
• 如果JSON字符串中有特殊字符（比如转义的引号、换行），一定要用正确的正则拆分，避免破坏JSON结构

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