上一篇解决了 Shuffle 的物理机制。这一篇进入内存管理——Spark 如何突破 JVM 的内存瓶颈。

Spark 是一个 JVM 应用，天然受制于 Java 的内存模型：对象头开销大、GC 停顿不可控、序列化效率低。Project Tungsten 是 Spark 为解决这三个问题发起的底层优化计划，它从三条线同时推进——堆外内存管理、二进制数据格式、全阶段代码生成。

本文只抓一个问题：Tungsten 的三条优化线分别解决了什么问题，以及 Spark 的统一内存管理模型如何在执行内存和存储内存之间做动态调配。

Java 对象的内存开销

一个 Java 字符串 “abcd” 在堆内占多少字节？

1
2
3
4
5
6
7
8
9
10
11
12
13
java.lang.String 对象:
  对象头:    12 bytes (64-bit JVM, 压缩指针)
  hash:      4 bytes (int)
  value[]:   4 bytes (引用)
  对齐填充:  4 bytes
  → 小计:   24 bytes

char[] 数组:
  对象头:    16 bytes (数组比普通对象多 4 bytes 存长度)
  数据:      8 bytes (4 个 char × 2 bytes)
  → 小计:   24 bytes

总计: 48 bytes，存储 4 个字符

一个只有 4 字节有效数据的字符串，实际占用 48 字节。对象头和元数据的开销是有效数据的 11 倍。在处理海量小记录时（比如 Shuffle 数据中的短 key），这个开销会显著膨胀 Spark 的内存消耗。

GC 问题同样严重。当 Executor 堆内存中有数百万个小对象时，Full GC 可能需要数秒甚至数十秒，期间 Executor 完全停顿，Task 执行被阻塞。

Tungsten 第一条线：堆外内存管理

Tungsten 的第一个策略是绕过 JVM 堆，直接使用堆外内存（Off-Heap Memory）。通过 sun.misc.Unsafe（或 Java 9+ 的 MemorySegment API），Spark 可以直接在操作系统的地址空间中分配和释放内存，不经过 JVM 的垃圾回收器。

Spark 的内存管理器（MemoryManager）把 Executor 的内存分成两个区域：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
Executor 内存布局（统一内存管理，自 Spark 1.6 起）:

  ┌─────────────────────────────────────────┐
  │              Reserved Memory            │  300MB 固定
  ├─────────────────────────────────────────┤
  │                                         │
  │         Execution Memory                │  用于 Shuffle、Join、Sort 的缓冲区
  │                                         │  ← 可以借用 Storage Memory 的空闲部分
  │─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ │  动态边界（默认各占 50%）
  │                                         │
  │         Storage Memory                  │  用于 RDD 缓存和广播变量
  │                                         │  ← 可以借用 Execution Memory 的空闲部分
  │                                         │
  ├─────────────────────────────────────────┤
  │           User Memory                   │  用户代码的 Java 对象
  └─────────────────────────────────────────┘

统一内存管理的关键设计是动态借用：当 Execution Memory 不够用而 Storage Memory 有空闲时，可以借用 Storage Memory 的空间（反之亦然）。但有一条不对称规则：Execution Memory 可以驱逐 Storage Memory 中的缓存数据来回收借出的空间，而 Storage Memory 不能驱逐 Execution Memory 的数据——因为执行中的数据（如 Shuffle 缓冲区）不能随意丢弃，而缓存数据可以通过 Lineage 重算。

Tungsten 第二条线：UnsafeRow 二进制格式

Tungsten 的第二个策略是用紧凑的二进制格式替代 Java 对象。Spark SQL 内部使用 UnsafeRow 作为行的统一表示：

1
2
3
4
5
6
7
8
9
10
UnsafeRow 内存布局（3 列: int, string, double）:

  ┌──────────┬──────────┬──────────┬──────────┬──────────┐
  │ null bits │ field 0  │ field 1  │ field 2  │ var data │
  │ (8 bytes) │ (8 bytes)│ (8 bytes)│ (8 bytes)│ (变长)   │
  └──────────┴──────────┴──────────┴──────────┴──────────┘
       │          │          │          │          │
       │       int 值      偏移+长度   double 值  字符串内容
       │       直接存储     指向 var    直接存储
       └─ 每个 bit 对应一列是否为 null

UnsafeRow 的优势：没有 Java 对象头，没有引用指针，没有 GC 压力。定长字段（int、long、double）直接内联存储，变长字段（string、binary）通过偏移量引用。整个行是一段连续的字节数组，可以直接在堆外内存中分配。

比较同一行数据的内存占用：

1
2
3
4
5
6
7
8
9
10
Java 对象表示: Row(42, "abcd", 3.14)
  Row 对象头:     24 bytes
  Integer 对象:   16 bytes
  String 对象:    48 bytes
  Double 对象:    24 bytes
  数组 + 引用:    ~40 bytes
  → 约 150 bytes

UnsafeRow 表示: [nullBits|42|offset+len|3.14|abcd]
  → 约 40 bytes

UnsafeRow 的内存效率大约是 Java 对象的 3-4 倍。在处理百万级行的 DataFrame 操作时，这个差异直接决定了能否在内存中完成计算。

Tungsten 第三条线：全阶段代码生成

Tungsten 的第三个策略是 Whole-Stage Code Generation（全阶段代码生成），它消除了传统查询引擎 Volcano 迭代器模型的虚函数调用开销。

传统 Volcano 模型中，每个算子实现一个 next() 方法，上层算子调用下层算子的 next() 逐条拉取数据：

1
2
3
4
5
6
7
8
Volcano 模型（每条记录经过多次虚函数调用）:

  Filter.next()
    └→ Project.next()
        └→ Scan.next()
            └→ 从文件读一行

  每处理一条记录：3 次虚函数调用 + 3 次方法栈帧

Whole-Stage CodeGen 把一个 Stage 内的所有算子融合成一个紧凑的循环：

1
2
3
4
5
6
7
8
9
CodeGen 后（编译成单个 Java 方法）:

  while (scan.hasNext()) {
    row = scan.next();
    if (row.getInt(0) > 10) {        // Filter 内联
      output.write(row.getInt(0),     // Project 内联
                   row.getString(1));
    }
  }

融合后没有虚函数调用、没有中间对象分配、没有方法栈帧切换。JIT 编译器可以进一步把整个循环优化成紧凑的机器码，充分利用 CPU 寄存器和缓存行。

在 Spark SQL 的 explain 输出中，带星号（*）的算子表示参与了 CodeGen：

1
2
*(1) Filter (id > 10)
+- *(1) Scan parquet

同一个括号内的数字相同表示这些算子被融合到同一个代码生成单元中。

实验：观察 CodeGen 的效果

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
val spark = SparkSession.builder.getOrCreate()
import spark.implicits._

// 准备测试数据
val df = spark.range(10000000).toDF("id")
  .withColumn("value", ($"id" % 100).cast("string"))

// CodeGen 开启（默认）
spark.conf.set("spark.sql.codegen.wholeStage", "true")
val t1 = System.nanoTime()
df.filter($"id" > 5000000).select($"id", $"value").count()
val elapsed1 = (System.nanoTime() - t1) / 1e6

// CodeGen 关闭
spark.conf.set("spark.sql.codegen.wholeStage", "false")
val t2 = System.nanoTime()
df.filter($"id" > 5000000).select($"id", $"value").count()
val elapsed2 = (System.nanoTime() - t2) / 1e6

println(s"CodeGen ON: ${elapsed1}ms, OFF: ${elapsed2}ms")

在典型硬件上，开启 CodeGen 的执行时间大约是关闭时的 1/2 到 1/5。差异取决于算子链的长度和数据量——算子越多、数据量越大，CodeGen 的收益越明显。

用 df.queryExecution.debug.codegen() 可以查看 Spark 为这条查询生成的 Java 代码。

模式提炼

Tungsten 的三条线体现了一个通用的性能优化思路：绕过高级抽象的开销，直接操作底层表示。

堆外内存绕过了 JVM 的 GC 机制。二进制格式绕过了 Java 的对象模型。代码生成绕过了面向对象的虚函数分派。每一条线都是在牺牲一定的开发便利性（手动内存管理、手动字节布局、运行时编译复杂度）换取执行效率。

这个思路在系统编程中反复出现。数据库引擎用 Buffer Pool 绕过 OS 的页缓存，用列式存储绕过行式对象模型，用向量化执行绕过逐条迭代。高频交易系统用 off-heap 的 ring buffer 绕过 GC，用内存映射绕过系统调用。

工程迁移表

Spark/Tungsten 概念 MapReduce 对应 Flink 对应 数据库对应
堆外内存管理 无（完全依赖 JVM 堆） Flink 的 MemorySegment（堆外） Buffer Pool（用户态内存管理）
UnsafeRow（二进制行格式） Writable 序列化 Flink 的 BinaryRow 数据库的行存储页格式（如 InnoDB 页）
Whole-Stage CodeGen 无 无（Flink 用向量化而非代码生成） 部分数据库的查询编译（如 HyPer、ClickHouse）
统一内存管理（动态借用） 固定的 Map/Reduce 缓冲区 Flink 的 Memory Manager（固定分配） 数据库的内存池管理（共享/排他区域）
溢写到磁盘 MapReduce 的 spill 机制 Flink 的 spill-to-disk 数据库的临时表空间

常见误解

误解一：“开启堆外内存就不会 OOM”。堆外内存不受 JVM 堆大小限制，但仍然受操作系统的物理内存和 Spark 配置的 spark.memory.offHeap.size 限制。如果堆外分配超过限制，Spark 会抛出 OutOfMemoryError 或触发溢写。

误解二：“Tungsten 只对 Spark SQL 有效”。UnsafeRow 和 CodeGen 确实主要服务于 Spark SQL / DataFrame API。但 Tungsten 的堆外内存管理和 Shuffle 优化（UnsafeShuffleWriter）也作用于 RDD API 的 Shuffle 操作。

误解三：“CodeGen 总是比 Volcano 模型快”。对于非常简单的查询（只有一个 Scan 算子），CodeGen 的编译开销可能超过执行节省的时间。Spark 会根据算子数量自动决定是否启用 CodeGen（默认阈值 spark.sql.codegen.maxFields=100）。

练习

  1. 在 spark-shell 中执行一个 DataFrame 查询，调用 df.queryExecution.debug.codegen() 查看生成的 Java 代码。找出代码中哪些部分对应 Filter 算子，哪些对应 Scan 算子。

  2. 配置 spark.memory.offHeap.enabled=true 和 spark.memory.offHeap.size=512m，执行一个大 Shuffle 作业。在 Spark UI 的 Executors 页面对比堆内和堆外内存的使用情况。

  3. 修改 spark.memory.storageFraction（默认 0.5）为 0.2，执行一个先 cache 再 join 的作业。观察 Storage Memory 被 Execution Memory 借用的情况。缓存的 RDD 是否被驱逐？

系列导航

序号 主题 状态
00 导读：为什么 Spark 的核心是一张 DAG 已完成
01 RDD：弹性分布式数据集的五大属性 已完成
02 宽依赖与窄依赖：Stage 是怎样划分出来的 已完成
03 DAG Scheduler 与 Task Scheduler：从逻辑计划到物理执行 已完成
04 Shuffle 机制：数据跨分区交换的代价与优化 已完成
05 内存管理与 Tungsten：堆外内存、序列化与代码生成 本篇
06 存储体系：Block Manager、广播变量与累加器 下一篇
07 容错机制：Lineage、Checkpoint 与推测执行
08 Spark SQL 与 Catalyst：从 SQL 文本到物理算子树
09 DataFrame 与 Dataset：类型安全与性能的折中
10 Adaptive Query Execution：运行时改写执行计划
11 数据源 API 与谓词下推：让存储层帮忙过滤
12 Structured Streaming：把流当成无界表
13 水位线与状态管理：流处理的时间语义
14 Spark on YARN / K8s：资源管理与部署模式
15 动态资源分配与作业调度：多租户集群的资源博弈
16 性能诊断：Spark UI、Event Log 与调优思路
17 Spark 与 Flink：两种计算引擎的设计选择

参考资料