Spark内存管理

本文基于Spark 1.6.0以后的版本
Spark 1.6.0引入了对堆外内存的管理并对内存管理模型进行了改进，SPARK-11389。

从物理上，分为堆内内存和堆外内存；从逻辑上分为execution内存和storage内存。
Execution内存主要是用来知足task执行过程当中某些算子对内存的需求，例如shuffle过程当中map端产生的中间结果须要缓存在内存中。
Storage内存主要用来存储RDD持久化的数据或者广播变量。

Off-heap内存

经过下面的代码片断(spark2.1版本)，能够清楚的知道execution内存和storage内存是如何分配Off-heap内存的。

protected[this] val maxOffHeapMemory = conf.getSizeAsBytes("spark.memory.offHeap.size", 0)
  protected[this] val offHeapStorageMemory =
    (maxOffHeapMemory * conf.getDouble("spark.memory.storageFraction", 0.5)).toLong

  offHeapExecutionMemoryPool.incrementPoolSize(maxOffHeapMemory - offHeapStorageMemory)
  offHeapStorageMemoryPool.incrementPoolSize(offHeapStorageMemory)

On-heap内存

对于on-heap内存的划分以下图

• 总内存
  spark2.1中经过下面的代码获取
  scala val systemMemory = conf.getLong("spark.testing.memory", Runtime.getRuntime.maxMemory)

• 系统预留内存

  预留内存在代码中是一个常量RESERVED_SYSTEM_MEMORY_BYTES指定为300M
  这里要求总内存至少是预留内存的1.5倍val minSystemMemory = (reservedMemory * 1.5).ceil.toLong
  而且会作以下的检测
  scala if (systemMemory < minSystemMemory) { throw new IllegalArgumentException(s"System memory $systemMemory must " + s"be at least $minSystemMemory. Please increase heap size using the --driver-memory " + s"option or spark.driver.memory in Spark configuration.") } // SPARK-12759 Check executor memory to fail fast if memory is insufficient if (conf.contains("spark.executor.memory")) { val executorMemory = conf.getSizeAsBytes("spark.executor.memory") if (executorMemory < minSystemMemory) { throw new IllegalArgumentException(s"Executor memory $executorMemory must be at least " + s"$minSystemMemory. Please increase executor memory using the " + s"--executor-memory option or spark.executor.memory in Spark configuration.") } }

• Spark可用内存

  Spark可用总内存=(系统内存-预留内存)*spark.memory.fraction
  val usableMemory = systemMemory - reservedMemory val memoryFraction = conf.getDouble("spark.memory.fraction", 0.6) (usableMemory * memoryFraction).toLong

• Storage内存
  Storage内存=Spark可用内存*spark.memory.storageFraction
  scala onHeapStorageRegionSize = (maxMemory * conf.getDouble("spark.memory.storageFraction", 0.5)).toLong

• Execution内存

  Execution内存=Spark可用内存-Storage内存

  private[spark] class UnifiedMemoryManager private[memory] (
    conf: SparkConf,
    val maxHeapMemory: Long,
    onHeapStorageRegionSize: Long,
    numCores: Int)
  extends MemoryManager(
    conf,
    numCores,
    onHeapStorageRegionSize,
    maxHeapMemory - onHeapStorageRegionSize)

• Storage内存与Execution内存的动态调整

  Storage can borrow as much execution memory as is free until execution reclaims its space. When this happens, cached blocks will be evicted from memory until sufficient borrowed memory is released to satisfy the execution memory request.

Similarly, execution can borrow as much storage memory as is free. However, execution memory is never evicted by storage due to the complexities involved in implementing this. The implication is that attempts to cache blocks may fail if execution has already eaten up most of the storage space, in which case the new blocks will be evicted immediately according to their respective storage levels.

上面这段文字是Spark官方对内存调整的注释，总结有以下几点
- 当execution内存有空闲的时候，storage能够借用execution的内存；当execution须要内存的时候， storage会释放借用的内存。这样作是安全的，由于storage内存若是不够能够溢出到本地磁盘。

- 当storage内存有空闲的时候也能够借给execution使用，可是当execution没有使用完的状况下是没法归还给storage的。由于execution是用来在计算过程当中存储临时结果的，若是内存被释放会致使后续的计算失败。

• user可支配内存

  这部份内存彻底由用户来支配，例如存储用户自定义的数据结构。

---

更多更好的文章请关注数客联盟