Spark 作為一個(gè)以擅長內(nèi)存計(jì)算為優(yōu)勢的計(jì)算引擎，內(nèi)存管理方案是其非常重要的模塊。作為使用者的我們，搞清楚 Spark 是如何管理內(nèi)存的，對我們編碼、調(diào)試及優(yōu)化過程會有很大幫助。本文之所以取名為 "Spark 新舊內(nèi)存管理方案剖析" 是因?yàn)樵?Spark 1.6 中引入了新的內(nèi)存管理方案，加之當(dāng)前很多公司還在使用 1.6 以前的版本，所以本文會對這兩種方案進(jìn)行剖析。

剛剛提到自 1.6 版本引入了新的內(nèi)存管理方案，但并不是說在 1.6 版本中不能使用舊的方案，而是默認(rèn)使用新方案。我們可以通過設(shè)置 spark.memory.userLegacyMode 值來選擇，該值為 false 表示使用新方案，true 表示使用舊方案，默認(rèn)為 false。該值是如何發(fā)揮作用的呢？看了下面的代碼就明白了：

val useLegacyMemoryManager = conf.getBoolean("spark.memory.useLegacyMode", false)
val memoryManager: MemoryManager =
  if (useLegacyMemoryManager) {
    new StaticMemoryManager(conf, numUsableCores)
  } else {
    UnifiedMemoryManager(conf, numUsableCores)
  }

根據(jù) spark.memory.useLegacyMode 值的不同，會創(chuàng)建 MemoryManager 不同子類的實(shí)例：

• 值為 false：創(chuàng)建 UnifiedMemoryManager 類實(shí)例，該類為新的內(nèi)存管理模塊的實(shí)現(xiàn)
• 值為 true：創(chuàng)建 StaticMemoryManager類實(shí)例，該類為舊的內(nèi)存管理模塊的實(shí)現(xiàn)

MemoryManager 是用于管理內(nèi)存的虛基類，聲明了一些方法來管理用于 execution 、 storage 的內(nèi)存和其他內(nèi)存：

• execution 內(nèi)存：用于 shuffles，如joins、sorts 和 aggregations，避免頻繁的 IO 而需要內(nèi)存 buffer
• storage 內(nèi)存：用于 caching RDD，緩存 broadcast 數(shù)據(jù)及緩存 task results
• 其他內(nèi)存：在下文中說明

先來看看 MemoryManager 重要的成員和方法：

接下來，來看看 MemoryManager 的兩種實(shí)現(xiàn)

StaticMemoryManager

當(dāng) spark.memory.userLegacyMode 為 true 時(shí)，在 SparkEnv 中是這樣實(shí)例化 StaticMemoryManager：

new StaticMemoryManager(conf, numUsableCores)

調(diào)用的是 StaticMemoryManager 輔助構(gòu)造函數(shù)，如下：

  def this(conf: SparkConf, numCores: Int) {
    this(
      conf,
      StaticMemoryManager.getMaxExecutionMemory(conf),
      StaticMemoryManager.getMaxStorageMemory(conf),
      numCores)
  }

繼而調(diào)用主構(gòu)造函數(shù)，如下：

private[spark] class StaticMemoryManager(
    conf: SparkConf,
    maxOnHeapExecutionMemory: Long,
    override val maxStorageMemory: Long,
    numCores: Int)
  extends MemoryManager(
    conf,
    numCores,
    maxStorageMemory,
    maxOnHeapExecutionMemory)

這樣我們就可以推導(dǎo)出，對于 StaticMemoryManager，其用于 storage 的內(nèi)存大小等于 StaticMemoryManager.getMaxStorageMemory(conf)；用于 execution 的內(nèi)存大小等于 StaticMemoryManager.getMaxExecutionMemory(conf)，下面進(jìn)一步看看這兩個(gè)方法的實(shí)現(xiàn)

StaticMemoryManager.getMaxExecutionMemory(conf)

實(shí)現(xiàn)如下：

  private def getMaxExecutionMemory(conf: SparkConf): Long = {
    val systemMaxMemory = conf.getLong("spark.testing.memory", Runtime.getRuntime.maxMemory)
    val memoryFraction = conf.getDouble("spark.shuffle.memoryFraction", 0.2)
    val safetyFraction = conf.getDouble("spark.shuffle.safetyFraction", 0.8)
    (systemMaxMemory * memoryFraction * safetyFraction).toLong
  }

若設(shè)置了 spark.testing.memory 則以該配置的值作為 systemMaxMemory，否則使用 JVM 最大內(nèi)存作為 systemMaxMemory。spark.testing.memory 僅用于測試，一般不設(shè)置，所以這里我們認(rèn)為 systemMaxMemory 的值就是 executor 的最大可用內(nèi)存。

spark.shuffle.memoryFraction：shuffle 期間用于 aggregation 和 cogroups 的內(nèi)存占 executor 運(yùn)行時(shí)內(nèi)存的百分比，用小數(shù)表示。在任何時(shí)候，用于 shuffle 的內(nèi)存總 size 不得超過這個(gè)限制，超出部分會 spill 到磁盤。如果經(jīng)常 spill，考慮調(diào)大 spark.storage.memoryFraction

spark.shuffle.safetyFraction：為防止 OOM，不能把 systemMaxMemory * spark.shuffle.memoryFraction 全用了，需要有個(gè)安全百分比

所以最終用于 execution 的內(nèi)存量為：executor 最大可用內(nèi)存 * spark.shuffle.memoryFraction * spark.shuffle.safetyFraction，默認(rèn)為 executor 最大可用內(nèi)存 * 0.16

需要特別注意的是，即使用于 execution 的內(nèi)存不夠用了，但同時(shí) executor 還有其他空余內(nèi)存，也不能給 execution 用

StaticMemoryManager.getMaxStorageMemory(conf)

實(shí)現(xiàn)如下：

  private def getMaxStorageMemory(conf: SparkConf): Long = {
    val systemMaxMemory = conf.getLong("spark.testing.memory", Runtime.getRuntime.maxMemory)
    val memoryFraction = conf.getDouble("spark.storage.memoryFraction", 0.6)
    val safetyFraction = conf.getDouble("spark.storage.safetyFraction", 0.9)
    (systemMaxMemory * memoryFraction * safetyFraction).toLong
  }

分析過程與 getMaxExecutionMemory 一致，我們得出這樣的結(jié)論，用于storage 的內(nèi)存量為: executor 最大可用內(nèi)存 * spark.storage.memoryFraction * spark.storage.safetyFraction，默認(rèn)為 executor 最大可用內(nèi)存 * 0.54

spark.storage.memoryFraction：用于做 memory cache 的內(nèi)存占 executor 最大可用內(nèi)存的百分比，該值不應(yīng)大于老生代

spark.storage.safetyFraction：防止 OOM 的安全比例，由 spark.storage.safetyFraction 控制，默認(rèn)為0.9。在 storage 中，有一部分內(nèi)存是給 unroll 使用的，unroll 即反序列化 block，該部分占比由 spark.storage.unrollFraction 控制，默認(rèn)為0.2

others

從上面的分析我們可以看到，storage 和 execution 總共使用了 80% 的內(nèi)存，那剩余 20% 去哪了？這部分內(nèi)存被系統(tǒng)保留了，用來存儲運(yùn)行中產(chǎn)生的對象

所以，各部分內(nèi)存占比可由下圖表示：

經(jīng)過上面的描述，我們搞明白了舊的內(nèi)存管理方案是如何劃分內(nèi)存的，也就可以根據(jù)我們實(shí)際的 app 來調(diào)整各個(gè)部分的比例。同時(shí)，我們可以明顯的看到這種內(nèi)存管理方式的缺陷，即 execution 和 storage 兩部分內(nèi)存固定死，不能共享，即使在一方內(nèi)存不夠用而另一方內(nèi)存空閑的情況下。這樣的方式經(jīng)常會造成內(nèi)存浪費(fèi)，所以有必要引入支持共享，能更好利用內(nèi)存的方案，UnifiedMemoryManager 就應(yīng)運(yùn)而生了

歡迎關(guān)注我的微信公眾號：FunnyBigData

FunnyBigData