PySpark处理8GB数组JSON展平转DataFrame遇2GB行大小限制问题

一、核心问题根源 #

你遇到的行大小超过2GB错误，本质是用multiline模式读取后，整个JSON被加载为单条记录，其中包含多个大数组字段（Table1、Table2）。当你对其中一个数组执行explode时，其他未被处理的大数组会被复制到每一行，导致单条记录的体积直接膨胀到原单条记录的大小（包含所有大数组），触发Gluten/Velox的行大小限制。

二、拆分表并展平数组（最优方案） #

不需要在读取时直接展平，而是通过拆分表+丢弃冗余字段的方式，避免大数组被重复复制。具体步骤如下：

1. 提前定义Schema（已知结构，避免推断开销）

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

val sourceSchema = StructType(Seq(
  StructField("Table1", ArrayType(StructType(Seq(
    StructField("col1", IntegerType),
    StructField("col2", StringType)
    // 补充Table1的其他字段
  )))),
  StructField("Table2", ArrayType(StructType(Seq(
    StructField("colA", StringType),
    StructField("colB", DoubleType)
    // 补充Table2的其他字段
  ))))
))

2. 读取整个JSON为单条记录

val rawDF = spark.read
  .schema(sourceSchema)
  .option("multiline", "true")
  .json("path/to/your/8gb.json")

3. 拆分每个表并展平数组
   处理每个表时，只提取目标数组字段，explode后展开嵌套结构，同时丢弃其他大数组字段，避免冗余数据：

// 处理Table1
val table1DF = rawDF
  .select(explode(col("Table1")).alias("table1_row"))
  .select("table1_row.*") // 展开Table1的所有字段

// 处理Table2
val table2DF = rawDF
  .select(explode(col("Table2")).alias("table2_row"))
  .select("table2_row.*") // 展开Table2的所有字段

这种方式下，explode后的每行仅包含单个表的单条数据，体积大幅缩小，不会触发行大小超限。

三、自定义UDF的注意事项 #

不推荐用自定义UDF优化读取效率，原因如下：

• UDF会绕过Spark Catalyst的优化逻辑，Gluten/Velox的列化加速也无法生效，反而会降低处理速度。
• 如果UDF需要处理大数组，依然可能遇到内存溢出问题，且调试难度更高。

如果一定要用UDF，优先选择Pandas UDF（Vectorized UDF），它比普通UDF性能更优，但依然不如原生explode+字段选择的组合高效。

四、额外优化建议 #

• 调整Spark内存配置：开启堆外内存spark.memory.offHeap.enabled=true，并设置合适的堆外内存大小spark.memory.offHeap.size，避免JVM内存压力。
• 调大分区数：设置spark.sql.shuffle.partitions为合理值（比如与CPU核心数匹配），让数据分散到更多分区处理。
• Gluten/Velox配置：调整spark.gluten.sql.columnar.batchsize为较小值（比如1024），减少单批处理的数据量，降低单条记录的内存占用。

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