如何拆分特殊算子Codegen解决Spark中Explode引发的OOM问题？

问题概述 #

使用多层explode函数的Spark任务出现内存溢出（OOM），分析显示多个explode被Codegen到同一个执行类中，中间结果在内存中大量缓冲，最终耗尽堆内存。

解决方案 #

1. 拆分Codegen（强制Stage分割） #

Spark的Codegen会将连续的算子合并到同一个Stage生成执行代码，要拆分Codegen，核心是在两次explode之间插入shuffle操作，强制Spark将任务拆分为多个Stage，让不同的explode在独立的Codegen类中执行，避免中间结果堆积。

具体操作：

• 在两次explode之间添加repartition（保持原分区数即可，无需改变并行度）
• 或使用persist(MEMORY_AND_DISK)，让中间结果落地磁盘，避免内存缓存

示例代码（Scala）：

// 优化前：连续explode导致Codegen合并
val originalDF = spark.table("source_table")
  .withColumn("col3", explode(split(coalesce($"table", lit("")), ":")))
  .withColumn("col7", explode(split(coalesce($"col1", lit("")), ":")))

// 优化后：插入repartition拆分Stage
val optimizedDF = spark.table("source_table")
  .withColumn("col3", explode(split(coalesce($"table", lit("")), ":")))
  .repartition(spark.sparkContext.defaultParallelism) // 触发Stage拆分，生成独立Codegen
  .withColumn("col7", explode(split(coalesce($"col1", lit("")), ":")))

2. 减少数据膨胀 #

• 提前过滤无效数据：在explode前过滤空字符串、拆分后无元素的字段，从源头减少数据量

val filteredDF = spark.table("source_table")
  .filter(length(coalesce($"table", lit(""))) > 0) // 过滤空值/空字符串
  .withColumn("col3", explode(split(coalesce($"table", lit("")), ":")))

• 合并拆分逻辑：如果业务允许，将多层拆分逻辑合并为单次处理，减少中间临时列的内存占用

3. 调整Spark内存配置 #

• 增大Executor堆内存：根据集群资源调整--executor-memory（例如16g）
• 优化内存比例：设置spark.memory.fraction=0.7（默认0.6），分配更多内存给执行计算
• 开启自适应执行：确保spark.sql.adaptive.enabled=true（默认开启），让Spark自动将溢出的中间结果写入磁盘

4. 兜底方案：禁用Generate算子Codegen #

如果以上方法无效，可以禁用GenerateExec的Codegen，强制Spark分批处理数据，但会牺牲性能：

spark.sql.codegen.excludeClasses=org.apache.spark.sql.execution.GenerateExec

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