4. 资源调优（下）：Spark内存模型详解（堆内与堆外）、内存压力测试方法、OOM的常见场景与解决方案

内存调优，说白了就是搞清楚你的数据到底在哪儿待着、怎么待着。很多同学上来就调 spark.executor.memory，结果该崩还是崩。我见过太多团队把内存配得很大，但任务跑着跑着就 OOM 了——原因很简单：他们根本不了解 Spark 的内存模型。

这一节，我们就把内存模型掰开揉碎。堆内、堆外、统一内存、压力测试、OOM 场景，一个一个来。

4.1 Spark 内存模型：堆内与堆外

Spark 的内存管理，从 JVM 视角看，分两大块：堆内内存（On-Heap）和堆外内存（Off-Heap）。

堆内就是 JVM 堆，受 GC 管理。堆外是直接向操作系统申请的内存，绕开了 JVM。为什么要搞堆外？因为 GC 在大堆场景下是个大坑——Full GC 一次可能几十秒，任务直接卡死。

Spark 1.6 之后引入了统一内存管理（Unified Memory Manager）。它把堆内内存划分成几个区域：

区域 默认比例 用途
Reserved Memory 300MB（固定） 系统预留，存储 Spark 内部对象
User Memory (可用内存 - Reserved) × 0.25 用户代码、数据结构、UDF 中间结果
Spark Memory (可用内存 - Reserved) × 0.75 执行（Shuffle、Join、Sort）和存储（Cache、Broadcast）共用
→ Execution Memory Spark Memory × 0.5（可抢占） 计算用，Shuffle 读写、聚合、排序
→ Storage Memory Spark Memory × 0.5（可被抢占） 缓存 RDD、DataFrame、Broadcast 变量

这里有个关键点：Execution 和 Storage 可以互相抢占。如果计算压力大，Execution 可以拿走 Storage 的内存；反过来，如果缓存数据多，Storage 也能占 Execution 的地盘。但注意——Storage 被抢占后，缓存的数据会被直接丢弃，需要重新计算。所以，如果你的任务重度依赖 Cache，最好调高 spark.memory.storageFraction，默认是 0.5。

堆外内存呢？它不受这些比例限制。堆外主要用于：

  • Netty 网络传输：Shuffle 读写时的零拷贝
  • 堆外缓存：通过 spark.memory.offHeap.enabled=true 开启，配合 spark.memory.offHeap.size 指定大小
  • PySpark / 原生库：Python UDF、TensorFlow 等

我个人习惯，在需要大量 Shuffle 或频繁 GC 的场景下，会开一点堆外内存。但别开太大——堆外内存泄漏很难排查，而且操作系统不会帮你回收。

核心公式：

Executor 可用内存 = spark.executor.memory - 300MB（Reserved）
Spark Memory = 可用内存 × 0.75
Execution Memory = Spark Memory × spark.memory.storageFraction（默认0.5）

举个例子：你配了 4GB executor，实际给计算用的 Execution Memory 只有 (4096 - 300) × 0.75 × 0.5 ≈ 1423MB。嗯，比你想象的小很多吧？

4.2 内存压力测试方法

光看配置不够，你得知道你的任务到底吃多少内存。我一般用三种方法做压力测试：

4.2.1 使用 Spark 内置的 Metrics

Spark UI 的 Stages 页面有个 Shuffle Read Size / Records 和 Shuffle Write Size / Records。这些能告诉你每个 Task 处理了多少数据。但更细粒度的内存使用，得看 Executors 页面的 Memory 列——它显示的是 Storage Memory 的使用量，不是 Execution。

想看到 Execution Memory 的实时使用？我建议用 Spark Listeners 自定义监控：

// 注册一个 SparkListener，监听 onExecutorMetricsUpdate 事件
// 可以拿到每个 Executor 的 onHeapExecutionMemory、offHeapExecutionMemory 等指标
// 然后推送到 Prometheus + Grafana

4.2.2 构造极端数据做压测

我曾经在一个实时数仓项目里，用 10 倍于线上峰值的数据量 做压测。具体做法：

  1. 准备一个大数据集（比如 100GB），用 repartition() 控制分区数
  2. 跑一个复杂的 Shuffle 操作（比如 groupBy().agg()）
  3. 观察 Spark UI 的 GC Time 和 Shuffle Spill
  4. 如果 GC Time 超过 Task 时间的 10%，或者有 Spill 到磁盘，说明内存不够

小技巧：用 spark.sql.adaptive.coalescePartitions.enabled=true 和 spark.sql.adaptive.advisoryPartitionSizeInBytes 控制输出分区大小，避免小文件过多。压测时建议关掉 AQE，先看原始内存压力。

4.2.3 使用 spark.memory.debug 和 JVM 工具

在 Spark 2.4+，可以开启 spark.memory.debug=true，它会在日志里打印每个 Task 的内存分配和释放情况。配合 jstat -gcutil 看 GC 频率，或者用 jmap -histo 看堆内对象分布。

嗯，这里要注意：生产环境别开 debug 日志，否则日志量会爆炸。我一般只在测试集群做压测时开。

4.3 OOM 的常见场景与解决方案

OOM 是 Spark 调优里最头疼的问题。我总结了几种高频场景：

场景 根因 典型表现 解决方案
Driver OOM collect() 拉回太多数据 Driver 端 GC 频繁，最后 OOM 改用 foreach() 或 take()；增大 spark.driver.memory
Executor OOM（Shuffle） 单个 Partition 数据倾斜 某个 Task 处理的数据量是其他 Task 的 10 倍以上 加盐、重分区、调整 spark.sql.adaptive.skewJoin.enabled
Executor OOM（Cache） Cache 数据超过 Storage Memory Storage Memory 使用率 100%，频繁驱逐 调大 spark.memory.storageFraction；改用序列化缓存 .persist(StorageLevel.MEMORY_AND_DISK_SER)
Executor OOM（Broadcast） Broadcast 变量太大 每个 Executor 都加载一个大对象 检查 spark.sql.autoBroadcastJoinThreshold，默认 10MB，调小或关闭
堆外 OOM Netty 缓冲区泄漏或 PySpark 内存泄漏 错误信息包含 OutOfDirectMemoryError 调大 spark.executor.extraJavaOptions=-XX:MaxDirectMemorySize；检查 UDF 是否有资源未释放

4.3.1 数据倾斜导致的 OOM

这是最常见的。我曾经遇到一个场景：一个 200 个 Partition 的 Join，其中一个 Partition 占了 80% 的数据。结果那个 Task 直接 OOM，其他 199 个 Task 几秒就跑完了。

解决方案：

  • 加盐：对倾斜的 Key 加随机前缀，打散到多个 Partition
  • AQE 自动倾斜处理：开启 spark.sql.adaptive.skewJoin.enabled，设置 spark.sql.adaptive.skewJoin.skewedPartitionFactor（默认 5）和 spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes（默认 256MB）
  • 手动重分区：df.repartition(1000, $"key") 或 df.coalesce(100)

避坑指南：我曾经在加盐时犯过一个错——加盐后忘记在 Join 条件里去掉盐值，导致 Join 结果全错了。加盐一定要在 Join 完成后 .drop("salt") 清理掉。

4.3.2 Cache 过多导致的 OOM

很多人喜欢把中间结果 .cache() 一下，觉得能加速。但如果你 Cache 的数据量超过了 Storage Memory，Spark 会开始驱逐旧数据。驱逐不掉？那就 OOM。

我的建议：

  • 能用 MEMORY_AND_DISK_SER 就别用 MEMORY_ONLY。序列化后内存占用能减少 2-5 倍
  • Cache 前先估算数据量：df.cache().count() 看 Storage Memory 使用量
  • 用完及时 .unpersist()

4.3.3 Broadcast 变量过大

Spark 默认会把小于 10MB 的表自动 Broadcast。但如果你的小表其实有 50MB，每个 Executor 都要加载一次，内存压力就上来了。

解决方案：

  • 调小 spark.sql.autoBroadcastJoinThreshold 到 5MB 或 1MB
  • 或者干脆关闭自动 Broadcast，手动控制：df.join(broadcast(smallDf), "key")

4.4 内存模型与 OOM 场景的关联图

下面这张图把堆内/堆外、各区域比例、以及 OOM 的触发点串起来了。你可以对照着看，哪个环节出问题，就调哪个参数。

Spark 内存模型与 OOM 场景关联图 Spark 内存模型与 OOM 场景关联图 Executor 总内存 (spark.executor.memory + spark.executor.memoryOverhead) 堆内内存 (On-Heap) Reserved Memory (300MB 固定) Spark 内部对象、系统预留 User Memory (可用内存 × 0.25) 用户代码、UDF 中间结果、自定义数据结构 Spark Memory (可用内存 × 0.75) Execution Memory Shuffle / Join / Sort / Agg 默认占 Spark Memory × 0.5 可抢占 Storage Storage Memory Cache / Broadcast / Accumulator 默认占 Spark Memory × 0.5 可被 Execution 抢占 抢占 堆外内存 (Off-Heap) Netty 传输 Shuffle 零拷贝 堆外缓存 spark.memory.offHeap.* PySpark / 原生库 Python UDF、C++ 库 常见 OOM 场景 ① Driver OOM (collect 过多) ② Executor OOM (数据倾斜) ③ Executor OOM (Cache 溢出) ④ Executor OOM (Broadcast 过大) ⑤ 堆外 OOM (DirectMemory) ⑥ GC 时间过长 (间接 OOM)

4.5 总结一下

内存调优没有银弹。你得先理解模型，再通过压测找到瓶颈，最后对症下药。我个人习惯的排查顺序：

  1. 看 Spark UI 的 GC Time 和 Shuffle Spill
  2. 检查是否有数据倾斜（看 Task 处理时间分布）
  3. 检查 Cache 和 Broadcast 的使用量
  4. 用 spark.memory.debug 或 JVM 工具定位具体区域
  5. 调整对应参数，再跑一轮压测验证

记住：内存不是越大越好。配得太大，GC 反而更频繁，任务更慢。找到那个“刚好够用”的点，才是真本事。

← 上一章 📖 返回目录 下一章 →