Spark 内存管理

朴灿烈づ我的快乐病毒、 2024-04-20 09:10 83阅读 0赞

#  #

# 6.Spark 内存管理 #

在执行 Spark 的应用程序时， Spark 集群会启动 Driver 和 Executor 两种

JVM 进程，前者为主控进程，负责创建 Spark 上下文，提交 Spark 作业（Job），

并将作业转化为计算任务（Task），在各个 Executor 进程间协调任务的调度，后

者负责在工作节点上执行具体的计算任务，并将结果返回给 Driver，同时为需要持

久化的 RDD 提供存储功能。由于 Driver 的内存管理相对来说较为简单，本节主

要对 Executor 的内存管理进行分析，下文中的 Spark 内存均特指 Executor 的内

存。

## 6.1 堆内和堆外内存规划 ##

作为一个 JVM 进程， Executor 的内存管理建立在 JVM 的内存管理之上，Spark 对

JVM 的堆内（On-heap）空间进行了更为详细的分配，以充分利用内存。同时， Spark 引入

了堆外（Off-heap）内存，使之可以直接在工作节点的系统内存中开辟空间，进一步优化了

内存的使用。

堆内内存受到 JVM 统一管理，堆外内存是直接向操作系统进行内存的申请和释放。

![1604514-20190710184800302-1842597328.png][]

**1. 堆内内存**

堆内内存的大小，由 Spark 应用程序启动时的 – executor-memory 或

spark.executor.memory 参数配置。Executor 内运行的并发任务共享 JVM 堆内内

存， 这些任务在缓存 RDD 数据和广播（Broadcast）数据时占用的内存被规划为存

储 （ Storage）内存 ， 而 这些任务在执行 Shuffle 时 占 用 的 内 存 被 规 划 为 执 行

（Execution）内存，剩余的部分不做特殊规划，那些 Spark 内部的对象实例，或者

用户定义的 Spark 应用程序中的对象实例，均占用剩余的空间。不同的管理模式下，

这三部分占用的空间大小各不相同。

Spark 对堆内内存的管理是一种逻辑上的”规划式”的管理，因为对象实例占

用内存的申请和释放都由 JVM 完成，Spark 只能在申请后和释放前记录这些内存，

我们来看其具体流程：

申请内存流程如下：

1. Spark 在代码中 new 一个对象实例；

2. JVM 从堆内内存分配空间，创建对象并返回对象引用；

3. Spark 保存该对象的引用，记录该对象占用的内存。

释放内存流程如下：

1. Spark 记录该对象释放的内存，删除该对象的引用；

2. 等待 JVM 的垃圾回收机制释放该对象占用的堆内内存。

我们知道，JVM 的对象可以以序列化的方式存储，序列化的过程是将对象转换

为二进制字节流，本质上可以理解为将非连续空间的链式存储转化为连续空间或块

存储，在访问时则需要进行序列化的逆过程——反序列化，将字节流转化为对象，

序列化的方式可以节省存储空间，但增加了存储和读取时候的计算开销。

对于 Spark 中序列化的对象，由于是字节流的形式，其占用的内存大小可直接

计算，而对于非序列化的对象，其占用的内存是通过周期性地采样近似估算而得，

即并不是每次新增的数据项都会计算一次占用的内存大小，这种方法降低了时间开

销但是有可能误差较大，导致某一时刻的实际内存有可能远远超出预期。此外， 在

被 Spark 标记为释放的对象实例，很有可能在实际上并没有被 JVM 回收，导致实

际可用的内存小于 Spark 记录的可用内存。所以 Spark 并不能准确记录实际可用

的堆内内存，从而也就无法完全避免内存溢出（OOM, Out of Memory）的异常。

虽然不能精准控制堆内内存的申请和释放，但 Spark 通过对存储内存和执行内

存各自独立的规划管理，可以决定是否要在存储内存里缓存新的 RDD，以及是否为

新的任务分配执行内存，在一定程度上可以提升内存的利用率，减少异常的出现。

**2. 堆外内存**

为了进一步优化内存的使用以及提高 Shuffle 时排序的效率， Spark 引入了堆

外（Off-heap）内存，使之可以直接在工作节点的系统内存中开辟空间，存储经过

序列化的二进制数据。

堆外内存意味着把内存对象分配在 Java 虚拟机的堆以外的内存，这些内存直接

受操作系统管理（而不是虚拟机）。这样做的结果就是能保持一个较小的堆，以减

少垃圾收集对应用的影响。

利用 JDK Unsafe API（从 Spark 2.0 开始，在管理堆外的存储内存时不再基于

Tachyon，而是与堆外的执行内存一样，基于 JDK Unsafe API 实现）， Spark 可以

直接操作系统堆外内存，减少了不必要的内存开销，以及频繁的 GC 扫描和回收，

提升了处理性能。堆外内存可以被精确地申请和释放（堆外内存之所以能够被精确

的申请和释放，是由于内存的申请和释放不再通过 JVM 机制，而是直接向操作系统

申请， JVM 对于内存的清理是无法准确指定时间点的，因此无法实现精确的释放），

而且序列化的数据占用的空间可以被精确计算，所以相比堆内内存来说降低了管理

的难度，也降低了误差。

在默认情况下堆外内存并不启用，可通过配置 spark.memory.offHeap.enabled 参

数启用，并由 spark.memory.offHeap.size 参数设定堆外空间的大小。 除了没有 other

空间，堆外内存与堆内内存的划分方式相同，所有运行中的并发任务共享存储内存

和执行内存。

（\*该部分内存主要用于程序的共享库、Perm Space、线程 Stack 和一些 Memory

mapping 等, 或者类 C 方式 allocate object）

## 6.2 内存空间分配 ##

**1. 静态内存管理**

在 Spark 最初采用的静态内存管理机制下， 存储内存、执行内存和其他内存的

大小在 Spark 应用程序运行期间均为固定的，但用户可以应用程序启动前进行配

置，堆内内存的分配如图 2 所示：

![1604514-20190710185243388-823980708.png][]

可以看到，可用的堆内内存的大小需要按照代码清单 1-1 的方式计算：

代码清单 1-1 堆内内存计算公式

可用的存储内存 = systemMaxMemory * spark.storage.memoryFraction * spark.storage.safetyFraction
    
    可用的执行内存 = systemMaxMemory * spark.shuffle.memoryFraction * spark.shuffle.safetyFraction

其中 systemMaxMemory 取决于当前 JVM 堆内内存的大小，最后可用的执行

内存或者存储内存要在此基础上与各自的 memoryFraction 参数和 safetyFraction

参数相乘得出。上述计算公式中的两个 safetyFraction 参数，其意义在于在逻辑上

预留出 1-safetyFraction 这么一块保险区域，降低因实际内存超出当前预设范围而

导致 OOM 的风险（上文提到，对于非序列化对象的内存采样估算会产生误差）。

值得注意的是，这个预留的保险区域仅仅是一种逻辑上的规划，在具体使用时 Spark

并没有区别对待，和”其它内存”一样交给了 JVM 去管理。

Storage 内存和 Execution 内存都有预留空间，目的是防止 OOM，因为 Spark 堆

内内存大小的记录是不准确的，需要留出保险区域。

堆外的空间分配较为简单，只有存储内存和执行内存，如图 1-3 所示。可用的

执行内存和存储内存占用的空间大小直接由参数 spark.memory.storageFraction 决

定， 由于堆外内存占用的空间可以被精确计算，所以无需再设定保险区域。

![1604514-20190710185533199-238447613.png][]

静态内存管理机制实现起来较为简单， 但如果用户不熟悉 Spark 的存储机制，

或没有根据具体的数据规模和计算任务或做相应的配置，很容易造成”一半海水，一

半火焰”的局面，即存储内存和执行内存中的一方剩余大量的空间，而另一方却早早

被占满，不得不淘汰或移出旧的内容以存储新的内容。由于新的内存管理机制的出

现，这种方式目前已经很少有开发者使用，出于兼容旧版本的应用程序的目的，Spark

仍然保留了它的实现。

**3. 统一内存管理**

Spark 1.6 之后引入的统一内存管理机制， 与静态内存管理的区别在于存储内存

和执行内存共享同一块空间，可以动态占用对方的空闲区域，统一内存管理的堆内

内存结构如图 1-4 所示：

![1604514-20190710185656514-2127070253.png][]

统一内存管理的堆外内存结构如图 1-5 所示：

![1604514-20190710185730837-872278285.png][]

其中最重要的优化在于动态占用机制，其规则如下：

1. 设定基本的存储内存和执行内存区域（spark.storage.storageFraction 参数），

该设定确定了双方各自拥有的空间的范围；

2. 双方的空间都不足时，则存储到硬盘；若己方空间不足而对方空余时，可

借用对方的空间;（存储空间不足是指不足以放下一个完整的 Block）

3. 执行内存的空间被对方占用后，可让对方将占用的部分转存到硬盘，然后”

归还”借用的空间；

4. 存储内存的空间被对方占用后，无法让对方”归还”，因为需要考虑 Shuffle

过程中的很多因素，实现起来较为复杂。

统一内存管理的动态占用机制如图 1-6 所示：

![1604514-20190710185852604-592248762.png][]

凭借统一内存管理机制，Spark 在一定程度上提高了堆内和堆外内存资源的利

用率，降低了开发者维护 Spark 内存的难度，但并不意味着开发者可以高枕无忧。

如果存储内存的空间太大或者说缓存的数据过多，反而会导致频繁的全量垃圾回收，

降低任务执行时的性能，因为缓存的 RDD 数据通常都是长期驻留内存的。所以要

想充分发挥 Spark 的性能，需要开发者进一步了解存储内存和执行内存各自的管理

方式和实现原理。

## 6.3 存储内存管理 ##

**1. RDD 的持久化机制**

弹性分布式数据集（RDD）作为 Spark 最根本的数据抽象，是只读的分区记录

（Partition）的集合，只能基于在稳定物理存储中的数据集上创建，或者在其他已有

的 RDD 上执行转换（Transformation）操作产生一个新的 RDD。转换后的 RDD 与

原始的 RDD 之间产生的依赖关系，构成了血统（Lineage）。 凭借血统，Spark 保

证了每一个 RDD 都可以被重新恢复。但 RDD 的所有转换都是惰性的，即只有当

一个返回结果给 Driver 的行动（Action）发生时，Spark 才会创建任务读取 RDD，

然后真正触发转换的执行。

Task 在启动之初读取一个分区时，会先判断这个分区是否已经被持久化，如果

没有则需要检查 Checkpoint 或按照血统重新计算。所以如果一个 RDD 上要执行

多次行动，可以在第一次行动中使用 persist 或 cache 方法，在内存或磁盘中持久

化或缓存这个 RDD，从而在后面的行动时提升计算速度。

事实上，cache 方法是使用默认的 MEMORY\_ONLY 的存储级别将 RDD 持久

化到内存，故缓存是一种特殊的持久化。 堆内和堆外存储内存的设计，便可以对缓

存 RDD 时使用的内存做统一的规划和管理。

RDD 的持久化由 Spark 的 Storage 模块负责，实现了 RDD 与物理存储的解

耦合。 Storage 模块负责管理 Spark 在计算过程中产生的数据，将那些在内存或磁

盘、在本地或远程存取数据的功能封装了起来。在具体实现时 Driver 端和 Executor

端 的 Storage 模 块 构 成 了 主 从 式 的 架 构 ， 即 Driver 端 的 BlockManager 为

Master，Executor 端的 BlockManager 为 Slave。

Storage 模块在逻辑上以 Block 为基本存储单位， RDD 的每个 Partition 经

过处理后唯一对应一个 Block（BlockId 的格式为 rdd\_RDD-ID\_PARTITION-ID ）。

Driver 端的 Master 负责整个 Spark 应用程序的 Block 的元数据信息的管理和维

护，而 Executor 端的 Slave 需要将 Block 的更新等状态上报到 Master，同时接收

Master 的命令，例如新增或删除一个 RDD。

![1604514-20190711175622167-1894274922.png][]

![1604514-20190710190121499-1664833181.png][]

在对 RDD 持久化时，Spark 规定了 MEMORY\_ONLY、MEMORY\_AND\_DISK

等 7 种不同的存储级别 ，而存储级别是以下 5 个变量的组合：

代码清单 5-1 resourceOffer代码

class StorageLevel private(
    　　private var _useDisk: Boolean, //磁盘
    　　private var _useMemory: Boolean, //这里其实是指堆内内存
    　　private var _useOffHeap: Boolean, //堆外内存
    　　private var _deserialized: Boolean, //是否为非序列化
    　　private var _replication: Int = 1 //副本个数
    )

Spark 中 7 种存储级别如下：

![1604514-20190710190424920-2072102481.png][]

![1604514-20190710190608614-1637670344.png][]

通 过 对 数 据 结 构 的 分 析 ， 可 以 看 出 存 储 级 别 从 三 个 维 度 定 义 了 RDD 的

Partition（同时也就是 Block）的存储方式：

1) ***存储位置***：磁盘／堆内内存／堆外内存。如 **MEMORY\_AND\_DISK** 是同时

在磁盘和堆内内存上存储，实现了冗余备份。 **OFF\_HEAP** 则是只在堆外内存存储，

目前选择堆外内存时不能同时存储到其他位置。

2) **存储形式** ： **Block** 缓 存 到 存 储 内 存 后 ， 是 否 为 非 序 列 化 的 形 式 。 如

**MEMORY\_ONLY** 是非序列化方式存储， **OFF\_HEAP** 是序列化方式存储。

3) **副本数量**：大于 1 时需要远程冗余备份到其他节点。如 **DISK\_ONLY\_2** 需

要远程备份 1 个副本。

**2. RDD 的缓存过程**

RDD 在缓存到存储内存之前，Partition 中的数据一般以迭代器（Iterator）的

数据结构来访问，这是 Scala 语言中一种遍历数据集合的方法。通过 Iterator 可以

获取分区中每一条序列化或者非序列化的数据项(Record)，这些 Record 的对象实例

在逻辑上 占用了 JVM 堆内内存 的 other 部 分的空间 ， 同 一 Partition 的 不同

Record 的存储空间并不连续。

RDD 在缓存到存储内存之后，Partition 被转换成 Block，Record 在堆内或堆

外存储内存中占用一块连续的空间。 将 Partition 由不连续的存储空间转换为连续存

储空间的过程，Spark 称之为"展开"（Unroll）。

Block 有序列化和非序列化两种存储格式，具体以哪种方式取决于该 RDD 的

存储级别。非序列化的 Block 以一种 DeserializedMemoryEntry 的数据结构定义，

用一个数组存储所有的对象实例，序列化的 Block 则以 SerializedMemoryEntry 的

数据结构定义，用字节缓冲区（ByteBuffer）来存储二进制数据。 每个 Executor 的

Storage 模块用一个链式 Map 结构（LinkedHashMap）来管理堆内和堆外存储内存

中所有的 Block 对象的实例，对这个 LinkedHashMap 新增和删除间接记录了内存

的申请和释放。

因为不能保证存储空间可以一次容纳 Iterator 中的所有数据，当前的计算任

务在 Unroll 时要向 MemoryManager 申请足够的 Unroll 空间来临时占位，空间

不足则 Unroll 失败，空间足够时可以继续进行。

对于序列化的 Partition，其所需的 Unroll 空间可以直接累加计算，一次申请。

对于非序列化的 Partition 则要在遍历 Record 的过程中依次申请，即每读取一

条 Record，采样估算其所需的 Unroll 空间并进行申请，空间不足时可以中断，释

放已占用的 Unroll 空间。

如果最终 Unroll 成功，当前 Partition 所占用的 Unroll 空间被转换为正常的

缓存 RDD 的存储空间，如下图所示。

![1604514-20190710190859155-1769689627.png][]

在静态内存管理时，Spark 在存储内存中专门划分了一块 Unroll 空间，其大

小是固定的， 统一内存管理时则没有对 Unroll 空间进行特别区分，当存储空间不

足时会根据动态占用机制进行处理。

**3. 淘汰与落盘**

由于同一个 Executor 的所有的计算任务共享有限的存储内存空间，当有新的

Block 需要缓存但是剩余空间不足且无法动态占用时，就要对 LinkedHashMap 中

的旧 Block 进行淘汰（Eviction），而被淘汰的 Block 如果其存储级别中同时包

含存储到磁盘的要求，则要对其进行落盘（Drop），否则直接删除该 Block。

存储内存的淘汰规则为：

 被淘汰的旧 Block 要与新 Block 的 MemoryMode 相同，即同属于堆外或

堆内内存；

 新旧 Block 不能属于同一个 RDD，避免循环淘汰；

 旧 Block 所属 RDD 不能处于被读状态，避免引发一致性问题；

 遍历 LinkedHashMap 中 Block， 按照最近最少使用（LRU）的顺序淘汰，

直到满足新 Block 所需的空间。其中 LRU 是 LinkedHashMap 的特性。

落盘的流程则比较简单，如果其存储级别符合\_useDisk 为 true 的条件，再根

据其\_deserialized 判断是否是非序列化的形式，若是则对其进行序列化，最后将数

据存储到磁盘，在 Storage 模块中更新其信息。

## 6.4 执行内存管理 ##

执行内存主要用来存储任务在执行 Shuffle 时占用的内存，Shuffle 是按照一定

规则对 RDD 数据重新分区的过程，我们来看 Shuffle 的 Write 和 Read 两阶段

对执行内存的使用：

** Shuffle Write**

1) 若在 map 端选择普通的排序方式，会采用 ExternalSorter 进行外排，在内

存中存储数据时主要占用堆内执行空间。

2) 若在 map 端选择 Tungsten 的排序方式，则采用 ShuffleExternalSorter 直

接对以序列化形式存储的数据排序，在内存中存储数据时可以占用堆外或堆内执行

空间，取决于用户是否开启了堆外内存以及堆外执行内存是否足够。

**　 Shuffle Read**

1) 在对 reduce 端的数据进行聚合时，要将数据交给 Aggregator 处理，在内

存中存储数据时占用堆内执行空间。

2) 如果需要进行最终结果排序，则要将再次将数据交给 ExternalSorter 处理，

占用堆内执行空间。

在 ExternalSorter 和 Aggregator 中，Spark 会 使用一种叫 AppendOnlyMap

的哈希表在堆内执行内存中存储数据，但在 Shuffle 过程中所有数据并不能都保存

到该哈希表中， 当这个哈希表占用的内存会进行周期性地采样估算，当其大到一定

程度，无法再从 MemoryManager 申请到新的执行内存时，Spark 就会将其全部内

容存储到磁盘文件中，这个过程被称为溢存(Spill)，溢存到磁盘的文件最后会被归

并(Merge)。

Shuffle Write 阶段中用到的 Tungsten 是 Databricks 公司提出的对 Spark 优

化内存和 CPU 使用的计划（钨丝计划），解决了一些 JVM 在性能上的限制和弊

端。Spark 会根据 Shuffle 的情况来自动选择是否采用 Tungsten 排序。

Tungsten 采用的页式内存管理机制建立在 MemoryManager 之上，即 Tungsten

对执行内存的使用进行了一步的抽象，这样在 Shuffle 过程中无需关心数据具体存

储在堆内还是堆外。

每个内存页用一个 MemoryBlock 来定义，并用 Object obj 和 long offset 这

两个变量统一标识一个内存页在系统内存中的地址。

堆内的 MemoryBlock 是以 long 型数组的形式分配的内存，其 obj 的值为是

这个数组的对象引用，offset 是 long 型数组的在 JVM 中的初始偏移地址，两者

配合使用可以定位这个数组在堆内的绝对地址；堆外的 MemoryBlock 是直接申请

到的内存块，其 obj 为 null，offset 是这个内存块在系统内存中的 64 位绝对地址。

Spark 用 MemoryBlock 巧 妙 地 将 堆 内和 堆外 内 存页 统一 抽 象封装 ， 并用 页表

(pageTable)管理每个 Task 申请到的内存页。

Tungsten 页式管理下的所有内存用 64 位的逻辑地址表示，由页号和页内偏移

量组成：

 页号：占 13 位，唯一标识一个内存页，Spark 在申请内存页之前要先申

请空闲页号。

 页内偏移量：占 51 位，是在使用内存页存储数据时，数据在页内的偏移

地址。

有了统一的寻址方式，Spark 可以用 64 位逻辑地址的指针定位到堆内或堆外

的内存，整个 Shuffle Write 排序的过程只需要对指针进行排序，并且无需反序列化，

整个过程非常高效，对于内存访问效率和 CPU 使用效率带来了明显的提升。

Spark 的存储内存和执行内存有着截然不同的管理方式： 对于存储内存来说，

Spark 用一个 LinkedHashMap 来集中管理所 有的 Block，Block 由需要缓存 的

RDD 的 Partition 转化而成；而对于执行内存，Spark 用 AppendOnlyMap 来存储

Shuffle 过程中的数据，在 Tungsten 排序中甚至抽象成为页式内存管理，开辟了全

新的 JVM 内存管理机制。

转载于:https://www.cnblogs.com/LXL616/p/11165976.html

[1604514-20190710184800302-1842597328.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/54e6f9bd9c3b41cd8b3b29cda16fd686.png
[1604514-20190710185243388-823980708.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/24d41c5fa6b046379e258e5cfbbec263.png
[1604514-20190710185533199-238447613.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/49ea98b482c04bd4b11abb74858b003c.png
[1604514-20190710185656514-2127070253.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/4b45776bc40346fcbbfb493524928941.png
[1604514-20190710185730837-872278285.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/e9c8d2b8da6f473ebc6c869115028d88.png
[1604514-20190710185852604-592248762.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/ecbcdda5b7a94ae9bac8c249564c133b.png
[1604514-20190711175622167-1894274922.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/65ed29742ec94e0f993ef51929b3d9cb.png
[1604514-20190710190121499-1664833181.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/cf6ca3bc347641ea804a2c801434d979.png
[1604514-20190710190424920-2072102481.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/324ed930623f4bd1bf93a6dacd55c7a2.png
[1604514-20190710190608614-1637670344.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/8864ab58167a4256a9413fabed215a32.png
[1604514-20190710190859155-1769689627.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/e1030bcdd6a34e7481a7a47dba2dd92b.png