spark 内存管理

2018-11-11  本文已影响0人  熊_看不见

内容目录

引言

       Spark 从1.6x开始对JVM的内存使用作出了一种全新的改变，Spark 1.6x以前是基于静态固定的JVM内存架构和运行机制。如果你不知道Spark到底对JVM怎么使用，就无法完全掌握和控制数据的缓存空间，所以理解Spark对JVM的内存使用是至关重要的。很多人对于Spark的印象是：它是基于内存的，而且可以缓存很多数据，显然Spark基于内存的观点是错误的，Spark只是优先充分地利用缓存。如果你不知道Spark可以缓存多少数据，就胡乱缓存数据的话，肯定会出问题。
       在数据规模已经确定的情况下，你有多少Executor和每个Executor分配多少内存(物理硬件资源确定的情况下)，你必须清楚知道你的内存最多能缓存多少数据；在shuffle过程中又使用了多少比例的缓存，这样对于算法的编写和业务的实现是至关重要的。
       文章会介绍Spark 2.x版本的内存使用比例，它被称为：Spark Unified Memory，这里的Unified是统一、联合的意思，Spark没有用Share这个词，是因为A和B进行Unified和A和B进行Share是完全不同的概念。Spark在运行过程中会出现不同类型的OOM，你必须搞清楚这个OOM背后是由于什么导致的。比如说我们使用算子mapPartition时候，一般会创建一些临时对象或者中间数据，你这个时候使用的临时对象和中间数据，是存储在一个UserSpace里面的用户操作空间，那你有没有想过这个空间的大小会导致应用程序出现OOM的情况，在Spark 2.x 中的Broadcast的数据存储在什么地方；ShuffleMapTask的数据又存储在什么地方。文章会介绍 JVM 在 Spark 1.6.X 以前和 2.X 版本对 Java 堆的使用，还会逐一解密上述几个疑问，也会简单介绍 Spark 1.6.x 以前版本在 Spark On Yarn 上内存的使用案例，希望这篇文章能为读者带出以下的启发：

<h2 id="1">JVM 內存使用架构剖析</h2>
    JVM的逻辑内存模型如下：


图片.png

     简单介绍一下各个部分：

     ==从 Spark 的角度来谈代码的运行和数据的处理，主要是谈 Java 堆 (Heap) 空间的运用。==

<h2 id="2">Spark 1.6.x以前版本内存管理</h2>

    下面两个图是Spark 1.6x以前版本对Java的堆(heap)的使用情况，左侧是Storage对内存的使用，右侧是Shuffle对内存的使用，这种方式叫做：StaticMemoryManagement，数据处理以及类的实体对象都放在JVM堆(heap)中。


图片.png

    JVM Heap默认情况下是512M，这是取决于spark.executor.memory的参数，在回答Spark JVM到底可以缓存多少数据之前，先了解一下JVM Heap在Spark中是如何分配内存比例的。无论你定义了Spark.executor.memory的内存空间有多大，Spark必然会定义一个安全空间，在默认的情况下只会使用JVM Heap的90%作为安全空间，在单个Executor的角度来说就是Heap size * 90%。

场景一：假设说在一个Executor，它可用的JVM Heap是10g，实际上Spark只能用90%，这个safe memory的的比例是由 ==spark.storage.safetyFraction== 控制的。(如果你单个的Executor的内存非常大，可以考虑提高这个比例)，在safe memory中也会划分为三个不同空间：Storage memory，Unroll memory、Shuffle memory。

也就是 Heap Size x 90% x 60%；Heap Size x 54%，在场景一的例子中是 10 x 0.9 x 0.6 = 5.4G；一个应用程序可以缓存多少数据是由 ==spark.storage.safetyFraction== 和 ==spark.storage.memoryFraction== 这两个参数共同决定的。

也就是 Heap Size x 90% x 60% x 20%；Heap Size x 10.8%，在场景一的例子中是 10 x 0.9 x 0.6 x 0.2 = 1.8G，你可能把序例化后的数据放在内存中，当你使用数据时，你需要把序例化的数据进行反序例化。
    对 cache 缓存数据的影响是由于 Unroll 是一个优先级较高的操作，进行 Unroll 操作的时候会占用 cache 的空间，而且又可以挤掉缓存在内存中的数据 (如果该数据的缓存级别是 MEMORY_ONLY 的话，否则该数据会丢失)。

    在 Shuffle 空间中也会有一个默认 80％ 的安全空间比例，所以应该是 Heap Size x 20% x 80%；Heap Size x 16%，在场景一的例子中是 10 x 0.2 x 0.8 = 1.6G。
    从内存的角度讲，你需要从远程抓取数据，抓取数据是一个 Shuffle 的过程，比如说你需要对数据进行排序，显现在这个过程中需要内存空间。

<h2 id="3">Spark on Yarn 计算内存使用案例</h2>
    这是一张 Spark 运行在 Yarn 上的架构图，它有 Driver 和 Executor 部份，在 Driver 部份有一个内存控制参数，Spark 1.6.x 以前是 spark.driver.memory，在实际生产环境下建义配置成 2G。如果 Driver 比较繁忙或者是经常把某些数据收集到 Driver 上的话，建义把这个参数调大一点。

    图的左边是 Executor 部份，它是被 Yarn 管理的，每台机制上都有一个 Node Manager；Node Manager 是被 Resources Manager 管理的，Resources Manager 的工作主要是管理全区级别的计算资源，计算资源核心就是内存和 CPU，每台机器上都有一个 Node Manager 来管理当前内存和 CPU 等资源。Yarn 一般跟 Hadoop 藕合，它底层会有 HDFS Node Manager，主要是负责管理当前机器进程上的数据并且与HDFS Name Node 进行通信。


图片.png

    在每个节点上至少有两个进程，一个是 HDFS Data Node，负责管理磁盘上的数据，另外一个是 Yarn Node Manager，负责管理执行进程，在这两个 Node 的下面有两个 Executors，每个 Executor 里面运行的都是 Tasks。从 Yarn 的角度来讲，会配置每个 Executor 所占用的空间，以防止资源竞争，Yarn 里有一个叫 Node Memory Pool 的概念，可以配置 64G 或者是 128G，Node Memory Pool 是当前节点上总共能够使用的内存大小。

    图中这两个 Executors 在两个不同的进程中 (JVM#1 和 JVM#2)，里面的 Task 是并行运行的，Task 是运行在线程中，但你可以配置 Task 使用线程的数量，e.g. 2条线程或者是4条线程，但默认情况下都是1条线程去处理一个Task，你也可以用 spark.executor.cores 去配置可用的 Core 以及 spark.executor.memory 去配置可用的 RAM 的大小。

在 Yarn 上启动 Spark Application 的时候可以通过以下参数来调优：

    场景一：例如 Yarn 集群上有 32 个 Node 来运行的 Node Manager，每个 Node 的内存是 64G，每个 Node 的 Cores 是 32 Cores。
    假如说每个 Node 我们要分配两个 Executors，那么可以把每个 Executor 分配 28G，Cores 分配为 12 个 Cores，每个 Spark Task 在运行的时候只需要一个 Core 就行啦，那么我们 32 个 Nodes 同时可以运行: 32 个 Node x 2 个 Executors x (12 个 Cores / 1) = 768 个 Task Slots，也就是说这个集群可以并行运行 768 个 Task，如果 Job 超过了 Task 可以并行运行的数量 (e.g. 768) 则需要排队。
    那么这个集群模可以缓存多少数据呢？从理论上：32 个 Node x 2 个 Executors x 28g x 90% 安全空间 x 60％缓存空间 = 967.68G，这个缓存数量对于普通的 Spark Job 而言是完全够用的，而实际上在运行中可能只能缓存 900G 的数据，900G 的数据从磁盘储存的角度数据有多大呢？还是 900G 吗？不是的，数据一般都会膨胀好几倍，这是和压缩、序列化和反序列化框架有关，所以在磁盘上可能也就 300G 的样子的数据。

<h2 id="4">Spark Unified Memory 的运行原理和机制(spark 1.6.x以后的内存管理器)</h2>
    下图是一种叫做联合内存(Spark Unified Memory)，数据缓存和数据执行直接的内存可以相互移动，这是一种更加弹性的方式。下图显示的是Spark 2.0.0 版本起JVM Heap的使用情况。


图片.png

==(1.6.x 到2.0.0之间的版本Spark Memory 占比是75%，User Memory 是25%)==

新型 JVM Heap 分成三个部份：Reserved Memory、User Memory 和 Spark Memory。

由两部分构成，分别是Storage Memory和Execution Memory。

现在 Storage 和 Execution (Shuffle) 采用了 Unified 的方式共同使用了 (Heap Size - 300MB) x 60%，默认情况下 Storage 和 Execution 各占该空间的 50%。
下图是 UnifiedMemoryManager.scala 中 UnifiedMemoryManager 伴生对象里的 apply 方法

图片.png

    定义：所谓 Unified 的意思是 Storgae 和 Execution 在适当时候可以借用彼此的 Memory，需要注意的是，当 Execution 空间不足而且 Storage 空间也不足的情况下，Storage 空间如果曾经使用了超过 Unified 默认的 50% 空间的话则超过部份会被强制 drop 掉一部份数据来解决 Execution 空间不足的问题 (注意：drop 后数据会不会丢失主要是看你在程序设置的 storage_level 来决定你是 Drop 到那里，可能 Drop 到磁盘上)，这是因为执行(Execution) 比缓存 (Storage) 是更重要的事情。

但是也有它的基本条件限制，Execution 向 Storage 借空间有两种情况：具体代码实现可以参考源码补充 : Spark 2.1.X 中 Unified 和 Static MemoryManager

下图是 Execution 向 Storage 借空间的第一种情况


图片.png

第一种情况：Storage 曾经向 Execution 借了空间，它缓存的数据可能是非常的多，然后 Execution 又不需要那么大的空间 (默认情况下各占 50%)，假设现在 Storage 占了 80%，Execution 占了 20%，然后 Execution 说自己空间不足，Execution 会向内存管理器发信号把 Storgae 曾经占用的超过 50％数据的那部份强制挤掉，在这个例子中挤掉了 30%；

下图是 Execution 向 Storage 借空间的第二种情况


图片.png

第二种情况：Execution 可以向 Storage Memory 借空间，在 Storage Memory 不足 50% 的情况下，Storgae Memory 会很乐意地把剩馀空间借给 Execution。相反当 Execution 有剩馀空间的时候，Storgae 也可以找 Execution 借空间。

上一篇下一篇

猜你喜欢

热点阅读