广州开发网站技术,wordpress数据库名和主机怎么填,信息造价网,泉州建设网站开发1、Spark内存管理
Spark 作为一个基于内存的分布式计算引擎#xff0c;其内存管理模块在整个系统中扮演着非常重要的角色。理解 Spark 内存管理的基本原理#xff0c;有助于更好地开发 Spark 应用程序和进行性能调优。本文中阐述的原理基于 Spark 2.1 版本。 在执行 Spark 的…1、Spark内存管理
Spark 作为一个基于内存的分布式计算引擎其内存管理模块在整个系统中扮演着非常重要的角色。理解 Spark 内存管理的基本原理有助于更好地开发 Spark 应用程序和进行性能调优。本文中阐述的原理基于 Spark 2.1 版本。 在执行 Spark 的应用程序时Spark 集群会启动 Driver 和 Executor 两种 JVM 进程前者为主控进程负责创建 Spark 上下文提交 Spark 作业Job并将作业转化为计算任务Task在各个 Executor 进程间协调任务的调度后者负责在工作节点上执行具体的计算任务并将结果返回给 Driver同时为需要持久化的 RDD 提供存储功能。由于 Driver 的内存管理相对来说较为简单本文主要对 Executor 的内存管理进行分析下文中的 Spark 内存均特指 Executor 的内存。
1.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 只能在申请后和释放前记录这些内存我们来看其具体流程 申请内存 1)Spark 在代码中 new 一个对象实例 2)JVM 从堆内内存分配空间创建对象并返回对象引用 3)Spark 保存该对象的引用记录该对象占用的内存 释放内存 1)Spark 记录该对象释放的内存删除该对象的引用 2)等待 JVM 的垃圾回收机制释放该对象占用的堆内内存 我们知道JVM 的对象可以以序列化的方式存储序列化的过程是将对象转换为二进制字节流本质上可以理解为将非连续空间的链式存储转化为连续空间或块存储在访问时则需要进行序列化的逆过程——反序列化将字节流转化为对象序列化的方式可以节省存储空间但增加了存储和读取时候的计算开销。 对于 Spark 中序列化的对象由于是字节流的形式其占用的内存大小可直接计算而对于非序列化的对象其占用的内存是通过周期性地采样近似估算而得即并不是每次新增的数据项都会计算一次占用的内存大小这种方法降低了时间开销但是有可能误差较大导致某一时刻的实际内存有可能远远超出预期[2]。此外在被 Spark 标记为释放的对象实例很有可能在实际上并没有被 JVM 回收导致实际可用的内存小于 Spark 记录的可用内存。所以 Spark 并不能准确记录实际可用的堆内内存从而也就无法完全避免内存溢出OOM, Out of Memory的异常。 虽然不能精准控制堆内内存的申请和释放但 Spark 通过对存储内存和执行内存各自独立的规划管理可以决定是否要在存储内存里缓存新的 RDD以及是否为新的任务分配执行内存在一定程度上可以提升内存的利用率减少异常的出现。
1.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.1.3内存管理接口
Spark 为存储内存和执行内存的管理提供了统一的接口——MemoryManager同一个 Executor 内的任务都调用这个接口的方法来申请或释放内存: 内存管理接口的主要方法 //申请存储内存 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 之后默认为统一管理Unified Memory Manager方式1.6 之前采用的静态管理Static Memory Manager方式仍被保留可通过配置 spark.memory.useLegacyMode 参数启用。两种方式的区别在于对空间分配的方式下面的第 2 小节会分别对这两种方式进行介绍。
1.2内存空间分配
1.2.1静态内存管理
在 Spark 最初采用的静态内存管理机制下存储内存、执行内存和其他内存的大小在 Spark 应用程序运行期间均为固定的但用户可以应用程序启动前进行配置堆内内存的分配如图 2 所示 图 2 . 静态内存管理图示——堆内 可以看到可用的堆内内存的大小需要按照下面的方式计算 可用堆内内存空间 1可用的存储内存 systemMaxMemory * spark.storage.memoryFraction * spark.storage.safetyFraction
2 可用的执行内存 systemMaxMemory * spark.shuffle.memoryFraction * spark.shuffle.safetyFraction 其中 systemMaxMemory 取决于当前 JVM 堆内内存的大小最后可用的执行内存或者存储内存要在此基础上与各自的 memoryFraction 参数和 safetyFraction 参数相乘得出。上述计算公式中的两个 safetyFraction 参数其意义在于在逻辑上预留出 1-safetyFraction 这么一块保险区域降低因实际内存超出当前预设范围而导致 OOM 的风险上文提到对于非序列化对象的内存采样估算会产生误差。值得注意的是这个预留的保险区域仅仅是一种逻辑上的规划在具体使用时 Spark 并没有区别对待和其它内存一样交给了 JVM 去管理。 堆外的空间分配较为简单只有存储内存和执行内存如图 3 所示。可用的执行内存和存储内存占用的空间大小直接由参数 spark.memory.storageFraction 决定由于堆外内存占用的空间可以被精确计算所以无需再设定保险区域。 图 3 . 静态内存管理图示——堆外 静态内存管理机制实现起来较为简单但如果用户不熟悉 Spark 的存储机制或没有根据具体的数据规模和计算任务或做相应的配置很容易造成一半海水一半火焰的局面即存储内存和执行内存中的一方剩余大量的空间而另一方却早早被占满不得不淘汰或移出旧的内容以存储新的内容。由于新的内存管理机制的出现这种方式目前已经很少有开发者使用出于兼容旧版本的应用程序的目的Spark 仍然保留了它的实现。
1.2.2统一内存管理
Spark 1.6 之后引入的统一内存管理机制与静态内存管理的区别在于存储内存和执行内存共享同一块空间可以动态占用对方的空闲区域如图 4 和图 5 所 图 4 . 统一内存管理图示——堆内 统一内存管理图示——堆外 其中最重要的优化在于动态占用机制其规则如下 1)设定基本的存储内存和执行内存区域spark.storage.storageFraction 参数该设定确定了双方各自拥有的空间的范围 2)双方的空间都不足时则存储到硬盘若己方空间不足而对方空余时可借用对方的空间;存储空间不足是指不足以放下一个完整的 Block 3)执行内存的空间被对方占用后可让对方将占用的部分转存到硬盘然后归还借用的空间 4)存储内存的空间被对方占用后无法让对方归还因为需要考虑 Shuffle 过程中的很多因素实现起来较为复杂 图 6 . 动态占用机制图示 凭借统一内存管理机制Spark 在一定程度上提高了堆内和堆外内存资源的利用率降低了开发者维护 Spark 内存的难度但并不意味着开发者可以高枕无忧。譬如所以如果存储内存的空间太大或者说缓存的数据过多反而会导致频繁的全量垃圾回收降低任务执行时的性能因为缓存的 RDD 数据通常都是长期驻留内存的 [5] 。所以要想充分发挥 Spark 的性能需要开发者进一步了解存储内存和执行内存各自的管理方式和实现原理。
1.3存储内存管理
1.3.1RDD 的持久化机制
弹性分布式数据集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 时使用的内存做统一的规划和管 理 存储内存的其他应用场景如缓存 broadcast 数据暂时不在本文的讨论范围之内。 RDD 的持久化由 Spark 的 Storage 模块 [7] 负责实现了 RDD 与物理存储的解耦合。Storage 模块负责管理 Spark 在计算过程中产生的数据将那些在内存或磁盘、在本地或远程存取数据的功能封装了起来。在具体实现时 Driver 端和 Executor 端的 Storage 模块构成了主从式的架构即 Driver 端的 BlockManager 为 MasterExecutor 端的 BlockManager 为 Slave。Storage 模块在逻辑上以 Block 为基本存储单位RDD 的每个 Partition 经过处理后唯一对应一个 BlockBlockId 的格式为 rdd_RDD-ID_PARTITION-ID 。Master 负责整个 Spark 应用程序的 Block 的元数据信息的管理和维护而 Slave 需要将 Block 的更新等状态上报到 Master同时接收 Master 的命令例如新增或删除一个 RDD。 图 7 . Storage 模块示意图 在对 RDD 持久化时Spark 规定了 MEMORY_ONLY、MEMORY_AND_DISK 等 7 种不同的 存储级别 而存储级别是以下 5 个变量的组合 清单 3 . 存储级别
class StorageLevel private(private var _useDisk: Boolean, //磁盘private var _useMemory: Boolean, //这里其实是指堆内内存private var _useOffHeap: Boolean, //堆外内存private var _deserialized: Boolean, //是否为非序列化private var _replication: Int 1 //副本个数
)通过对数据结构的分析可以看出存储级别从三个维度定义了 RDD 的 Partition同时也就是 Block的存储方式 1)存储位置磁盘堆内内存堆外内存。如 MEMORY_AND_DISK 是同时在磁盘和堆内内存上存储实现了冗余备份。OFF_HEAP 则是只在堆外内存存储目前选择堆外内存时不能同时存储到其他位置。 2)存储形式Block 缓存到存储内存后是否为非序列化的形式。如 MEMORY_ONLY 是非序列化方式存储OFF_HEAP 是序列化方式存储。 3)副本数量大于 1 时需要远程冗余备份到其他节点。如 DISK_ONLY_2 需要远程备份 1 个副本。
1.3.2RDD 缓存的过程
RDD 在缓存到存储内存之前Partition 中的数据一般以迭代器Iterator的数据结构来访问这是 Scala 语言中一种遍历数据集合的方法。通过 Iterator 可以获取分区中每一条序列化或者非序列化的数据项(Record)这些 Record 的对象实例在逻辑上占用了 JVM 堆内内存的 other 部分的空间同一 Partition 的不同 Record 的空间并不连续。 RDD 在缓存到存储内存之后Partition 被转换成 BlockRecord 在堆内或堆外存储内存中占用一块连续的空间。将Partition由不连续的存储空间转换为连续存储空间的过程Spark称之为展开Unroll。Block 有序列化和非序列化两种存储格式具体以哪种方式取决于该 RDD 的存储级别。非序列化的 Block 以一种 DeserializedMemoryEntry 的数据结构定义用一个数组存储所有的对象实例序列化的 Block 则以 SerializedMemoryEntry的数据结构定义用字节缓冲区ByteBuffer来存储二进制数据。每个 Executor 的 Storage 模块用一个链式 Map 结构LinkedHashMap来管理堆内和堆外存储内存中所有的 Block 对象的实例[6]对这个 LinkedHashMap 新增和删除间接记录了内存的申请和释放。 因为不能保证存储空间可以一次容纳 Iterator 中的所有数据当前的计算任务在 Unroll 时要向 MemoryManager 申请足够的 Unroll 空间来临时占位空间不足则 Unroll 失败空间足够时可以继续进行。对于序列化的 Partition其所需的 Unroll 空间可以直接累加计算一次申请。而非序列化的 Partition 则要在遍历 Record 的过程中依次申请即每读取一条 Record采样估算其所需的 Unroll 空间并进行申请空间不足时可以中断释放已占用的 Unroll 空间。如果最终 Unroll 成功当前 Partition 所占用的 Unroll 空间被转换为正常的缓存 RDD 的存储空间如下图 8 所示。 图 8. Spark Unroll 示意图 在图 3 和图 5 中可以看到在静态内存管理时Spark 在存储内存中专门划分了一块 Unroll 空间其大小是固定的统一内存管理时则没有对 Unroll 空间进行特别区分当存储空间不足时会根据动态占用机制进行处理。
1.3.3淘汰和落盘
由于同一个 Executor 的所有的计算任务共享有限的存储内存空间当有新的 Block 需要缓存但是剩余空间不足且无法动态占用时就要对 LinkedHashMap 中的旧 Block 进行淘汰Eviction而被淘汰的 Block 如果其存储级别中同时包含存储到磁盘的要求则要对其进行落盘Drop否则直接删除该 Block。 存储内存的淘汰规则为 1)被淘汰的旧 Block 要与新 Block 的 MemoryMode 相同即同属于堆外或堆内内存 2)新旧 Block 不能属于同一个 RDD避免循环淘汰 3)旧 Block 所属 RDD 不能处于被读状态避免引发一致性问题 4)遍历 LinkedHashMap 中 Block按照最近最少使用LRU的顺序淘汰直到满足新 Block 所需的空间。其中 LRU 是 LinkedHashMap 的特性。 落盘的流程则比较简单如果其存储级别符合_useDisk 为 true 的条件再根据其_deserialized 判断是否是非序列化的形式若是则对其进行序列化最后将数据存储到磁盘在 Storage 模块中更新其信息。
1.4执行内存管理
1.4.1多任务间内存分配
Executor 内运行的任务同样共享执行内存Spark 用一个 HashMap 结构保存了任务到内存耗费的映射。每个任务可占用的执行内存大小的范围为 1/2N ~ 1/N其中 N 为当前 Executor 内正在运行的任务的个数。每个任务在启动之时要向 MemoryManager 请求申请最少为 1/2N 的执行内存如果不能被满足要求则该任务被阻塞直到有其他任务释放了足够的执行内存该任务才可以被唤醒。
1.4.2Shuffle 的内存占用
执行内存主要用来存储任务在执行 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 使用的计划[9]解决了一些 JVM 在性能上的限制和弊端。Spark 会根据 Shuffle 的情况来自动选择是否采用 Tungsten 排序。Tungsten 采用的页式内存管理机制建立在 MemoryManager 之上即 Tungsten 对执行内存的使用进行了一步的抽象这样在 Shuffle 过程中无需关心数据具体存储在堆内还是堆外。每个内存页用一个 MemoryBlock 来定义并用 Object obj 和 long offset 这两个变量统一标识一个内存页在系统内存中的地址。堆内的 MemoryBlock 是以 long 型数组的形式分配的内存其 obj 的值为是这个数组的对象引用offset 是 long 型数组的在 JVM 中的初始偏移地址两者配合使用可以定位这个数组在堆内的绝对地址堆外的 MemoryBlock 是直接申请到的内存块其 obj 为 nulloffset 是这个内存块在系统内存中的 64 位绝对地址。Spark 用 MemoryBlock 巧妙地将堆内和堆外内存页统一抽象封装并用页表(pageTable)管理每个 Task 申请到的内存页。 Tungsten 页式管理下的所有内存用 64 位的逻辑地址表示由页号和页内偏移量组成 页号占 13 位唯一标识一个内存页Spark 在申请内存页之前要先申请空闲页号。 页内偏移量占 51 位是在使用内存页存储数据时数据在页内的偏移地址。 有了统一的寻址方式Spark 可以用 64 位逻辑地址的指针定位到堆内或堆外的内存整个 Shuffle Write 排序的过程只需要对指针进行排序并且无需反序列化整个过程非常高效对于内存访问效率和 CPU 使用效率带来了明显的提升[10]。 Spark 的存储内存和执行内存有着截然不同的管理方式对于存储内存来说Spark 用一个 LinkedHashMap 来集中管理所有的 BlockBlock 由需要缓存的 RDD 的 Partition 转化而成而对于执行内存Spark 用 AppendOnlyMap 来存储 Shuffle 过程中的数据在 Tungsten 排序中甚至抽象成为页式内存管理开辟了全新的 JVM 内存管理机制。
