Apache Spark 内存管理详解（1）

look_w

Spark 作为一个基于内存的分布式计算引擎，其内存管理模块在整个系统中扮演着非常重要的角色。理解 Spark 内存管理的基本原理，有助于更好地开发                Spark 应用程序和进行性能调优。本文旨在梳理出 Spark 内存管理的脉络，抛砖引玉，引出读者对这个话题的深入探讨。本文中阐述的原理基于                Spark 2.1 版本，阅读本文需要读者有一定的 Spark 和 Java 基础，了解 RDD、Shuffle、JVM 等相关概念。
在执行 Spark 的应用程序时，Spark 集群会启动 Driver 和 Executor 两种 JVM 进程，前者为主控进程，负责创建 Spark                上下文，提交 Spark 作业（Job），并将作业转化为计算任务（Task），在各个 Executor                进程间协调任务的调度，后者负责在工作节点上执行具体的计算任务，并将结果返回给 Driver，同时为需要持久化的 RDD                提供存储功能[1]。由于 Driver 的内存管理相对来说较为简单，本文主要对 Executor 的内存管理进行分析，下文中的                Spark 内存均特指 Executor 的内存。
1.                堆内和堆外内存规划作为一个 JVM 进程，Executor 的内存管理建立在 JVM 的内存管理之上，Spark 对 JVM                的堆内（On-heap）空间进行了更为详细的分配，以充分利用内存。同时，Spark                引入了堆外（Off-heap）内存，使之可以直接在工作节点的系统内存中开辟空间，进一步优化了内存的使用。
图 1 .                    堆内和堆外内存示意图1.1 堆内内存堆内内存的大小，由 Spark 应用程序启动时的 –executor-memory 或 spark.executor.memory                参数配置。Executor 内运行的并发任务共享 JVM 堆内内存，这些任务在缓存 RDD                数据和广播（Broadcast）数据时占用的内存被规划为存储（Storage）内存，而这些任务在执行 Shuffle                时占用的内存被规划为执行（Execution）内存，剩余的部分不做特殊规划，那些 Spark 内部的对象实例，或者用户定义的 Spark                应用程序中的对象实例，均占用剩余的空间。不同的管理模式下，这三部分占用的空间大小各不相同（下面第 2 小节会进行介绍）。
Spark 对堆内内存的管理是一种逻辑上的"规划式"的管理，因为对象实例占用内存的申请和释放都由 JVM 完成，Spark                    只能在申请后和释放前记录这些内存，我们来看其具体流程：

• 申请内存：

• Spark 在代码中 new 一个对象实例
• JVM 从堆内内存分配空间，创建对象并返回对象引用
• Spark 保存该对象的引用，记录该对象占用的内存

• 释放内存：

• Spark 记录该对象释放的内存，删除该对象的引用
• 等待 JVM 的垃圾回收机制释放该对象占用的堆内内存

我们知道，JVM                的对象可以以序列化的方式存储，序列化的过程是将对象转换为二进制字节流，本质上可以理解为将非连续空间的链式存储转化为连续空间或块存储，在访问时则需要进行序列化的逆过程——反序列化，将字节流转化为对象，序列化的方式可以节省存储空间，但增加了存储和读取时候的计算开销。
对于 Spark                中序列化的对象，由于是字节流的形式，其占用的内存大小可直接计算，而对于非序列化的对象，其占用的内存是通过周期性地采样近似估算而得，即并不是每次新增的数据项都会计算一次占用的内存大小，这种方法降低了时间开销但是有可能误差较大，导致某一时刻的实际内存有可能远远超出预期[2]。此外，在被                Spark 标记为释放的对象实例，很有可能在实际上并没有被 JVM 回收，导致实际可用的内存小于 Spark 记录的可用内存。所以 Spark                并不能准确记录实际可用的堆内内存，从而也就无法完全避免内存溢出（OOM, Out of Memory）的异常。
虽然不能精准控制堆内内存的申请和释放，但 Spark 通过对存储内存和执行内存各自独立的规划管理，可以决定是否要在存储内存里缓存新的                RDD，以及是否为新的任务分配执行内存，在一定程度上可以提升内存的利用率，减少异常的出现。
1.2 堆外内存为了进一步优化内存的使用以及提高 Shuffle 时排序的效率，Spark                引入了堆外（Off-heap）内存，使之可以直接在工作节点的系统内存中开辟空间，存储经过序列化的二进制数据。利用 JDK Unsafe API（从                Spark 2.0 开始，在管理堆外的存储内存时不再基于 Tachyon，而是与堆外的执行内存一样，基于 JDK Unsafe API                实现[3]），Spark 可以直接操作系统堆外内存，减少了不必要的内存开销，以及频繁的 GC                扫描和回收，提升了处理性能。堆外内存可以被精确地申请和释放，而且序列化的数据占用的空间可以被精确计算，所以相比堆内内存来说降低了管理的难度，也降低了误差。
在默认情况下堆外内存并不启用，可通过配置 spark.memory.offHeap.enabled 参数启用，并由                spark.memory.offHeap.size 参数设定堆外空间的大小。除了没有 other                空间，堆外内存与堆内内存的划分方式相同，所有运行中的并发任务共享存储内存和执行内存。
1.3 内存管理接口Spark 为存储内存和执行内存的管理提供了统一的接口——MemoryManager，同一个 Executor                内的任务都调用这个接口的方法来申请或释放内存:
清单 1 .                内存管理接口的主要方法

1 2 3 4 5 6 7 8 9 10 11 12

//申请存储内存 def acquireStorageMemory(blockId: BlockId, numBytes: Long, memoryMode: MemoryMode): Boolean //申请展开内存 def acquireUnrollMemory(blockId: BlockId, numBytes: Long, memoryMode: MemoryMode): Boolean //申请执行内存 def acquireExecutionMemory(numBytes: Long, taskAttemptId: Long, memoryMode: MemoryMode): Long //释放存储内存 def releaseStorageMemory(numBytes: Long, memoryMode: MemoryMode): Unit //释放执行内存 def releaseExecutionMemory(numBytes: Long, taskAttemptId: Long, memoryMode: MemoryMode): Unit //释放展开内存 def releaseUnrollMemory(numBytes: Long, memoryMode: MemoryMode): Unit

我们看到，在调用这些方法时都需要指定其内存模式（MemoryMode），这个参数决定了是在堆内还是堆外完成这次操作。
MemoryManager 的具体实现上，Spark 1.6 之后默认为统一管理（）方式，1.6 之前采用的静态管理（）方式仍被保留，可通过配置 spark.memory.useLegacyMode                参数启用。两种方式的区别在于对空间分配的方式，下面的第 2 小节会分别对这两种方式进行介绍。