从Gzip压缩JSON数组文件读取到Spark DataFrame的性能问题

我之前也碰到过一模一样的场景——手里只有这种整个文件是一个大JSON数组的Gzip压缩包，小文件跑起来飞快，GB级大文件直接卡到怀疑人生。其实核心原因有两个：一是Gzip是不可拆分的压缩格式，Spark没法把大文件拆成多个块并行处理；二是Spark默认的JSON Reader期望每行一个JSON对象，而不是整个文件塞一个大数组，导致解析时只能单线程处理整个文件。

下面给你几个可行的解决办法，按是否能预处理文件来分：

一、无法预处理文件时的实时处理方案 #

1. 使用wholeTextFiles读取+JSON数组解析+展开 #

这个方法先把整个Gzip文件的内容读成字符串，再解析成数组，最后展开成单个对象，这样后续的计算就能并行处理了：

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

// 先定义对应的Schema，避免Spark自动推断浪费时间
val productSchema = StructType(Seq(
  StructField("id", IntegerType),
  StructField("image", StringType)
))

val itemSchema = StructType(Seq(
  StructField("Product", productSchema),
  StructField("Color", StringType)
))

val arraySchema = ArrayType(itemSchema)

// 读取所有Gzip文件，每个文件对应一条记录
val rawDf = spark.sparkContext.wholeTextFiles("/path/to/*.gz").toDF("file_path", "content")

// 解析JSON数组并展开成单个Item行
val finalDf = rawDf
  .select(from_json(col("content"), arraySchema).alias("items"))
  .select(explode(col("items")).alias("item"))
  .select("item.Product.*", "item.Color") // 提取需要的字段

2. 用Jackson流式解析优化性能 #

如果觉得Spark内置的JSON解析不够快，可以直接用Jackson（Spark底层也是用它）做流式解析，在mapPartitions里处理每个文件的内容，效率会更高：

import com.fasterxml.jackson.databind.ObjectMapper
import com.fasterxml.jackson.module.scala.DefaultScalaModule
import scala.collection.JavaConverters._

// 定义样例类映射JSON结构
case class Product(id: Int, image: String)
case class Item(Product: Product, Color: String)

// 初始化Jackson的ObjectMapper，注册Scala模块
val mapper = new ObjectMapper()
mapper.registerModule(DefaultScalaModule)

// 读取文件并解析数组
val itemsRdd = spark.sparkContext.wholeTextFiles("/path/to/*.gz")
  .flatMap { case (_, content) =>
    // 把整个字符串解析成Item数组，再转成List
    mapper.readValue(content, classOf[Array[Item]]).toList
  }

// 转换成DataFrame
val finalDf = itemsRdd.toDF()

注意：如果单个Gzip文件特别大（比如几十GB），这种方法还是会因为单分区解析而变慢，这时候最好还是想办法预处理文件。

二、可以预处理文件时的最优方案 #

如果有机会提前处理文件，强烈建议把它转换成Spark友好的格式，步骤如下：

1. 解压Gzip文件：把大数组的JSON文件解压出来。
2. 拆分JSON数组：把[{"a":1},{"b":2}]这种格式转换成每行一个JSON对象：

   {"a":1}
   {"b":2}
   

3. 用可拆分的压缩格式重新压缩：比如用Snappy、Bzip2（不要用Gzip了），这些格式支持Spark并行读取文件块。

举个Python预处理脚本的例子：

import json
import gzip
import snappy

# 读取Gzip里的JSON数组
with gzip.open("large_input.gz", "rt", encoding="utf-8") as f:
    data = json.load(f)

# 拆分并压缩成Snappy格式
with open("processed_output.json.snappy", "wb") as f:
    for item in data:
        line = json.dumps(item) + "\n"
        f.write(snappy.compress(line.encode("utf-8")))

之后用Spark读取就快得飞起了：

val finalDf = spark.read.json("/path/to/processed_output.json.snappy")

额外优化小技巧 #

• 调整Spark并行度：解析完成后可以用finalDf.repartition(100)（根据集群规模调整数量），让后续计算充分并行。
• 增加Executor内存：解析大JSON数组需要较多内存，在提交任务时可以设置--executor-memory 16G这类参数。
• 关闭Schema自动推断：提前定义好Schema，避免Spark扫描整个文件推断Schema浪费时间。

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