spark Executor的内存大小设置一直是长期困扰开发人员和分析师的一个大问题。不同应用使用的算法和数据不同，一次内存设置也是难以评估，设置过大会造成资源浪费，其余任务得不到资源而等待。设置过小，计算过程中会带来频繁的GC和磁盘读写，缓存数据比较少，还有可能会带来OOM，也会影响到性能。
针对上述的问题，spark 1.6 带来了新的内存管理机制，Unified Memory Manager。
以前的内存管理
日常使用中，我们通过spark.executor.memory来控制一个executor最多可以使用的内存大小，实际上是通过设置Executor的JVM的Heap大小实现的。
在spark1.6之前，Executor的内存界限分明，分别由3部分组成：execution,storage和system。
execution
execution空间通过设置spark.shuffle.memoryFraction参数来控制大小，默认为0.2。为了避免shuffle，join，排序和聚合这些操作直接将数据写入磁盘，所设置的buffer大小，减少了磁盘读写的次数。storage
storage空间通过设置spark.storage.memoryFraction参数来控制大小，默认为0.6。用于存储用户显示调用的persist,cache,broadcast等命令存储的数据空间。system
程序运行需要的空间，存储一些spark内部的元数据信息，用户的数据结构，避免一些不寻常的大记录带来的OOM。
之前的管理方式，最明显的就是对execution和storage空间进行了明显的划分。举个例子，一些任务可能对数据缓存的需求并不是很高，就会造成storage空间的浪费。
因此，spark1.6带来了新的内存管理机制。
Unified MemoryManager简述
为了解决上述出现的问题，新提出的内存管理机制淡化了execution空间和storage空间的边界，让它们之间可以相互“借”内存。
通过spark.memory.useLegacyMode参数来决定是否开启新的内存管理机制，默认为开启。
它们总共可用的内存由spark.memory.fraction参数控制（默认为0.75）。在该空间内部，对execution和storage进行了进一步的划分。由spark.memory.storageFraction参数控制（默认为0.5），这意味着strorage空间真实占有的executor空间为：
0.75*0.5 = 0.375
exetution向storage”借“内存
在程序执行过程中，如果execution空间不足，则会向storage空间提出申请，storage会将空闲内存借给execution使用，如果不够，则会释放之前向execution借的内存。
execution申请在acquireStorageMemory方法中实现。
每次申请空间前会进行相应判断，在maybeGrowExecutionPool方法中:
//extraMemoryNeeded为需要申请的内存
def maybeGrowExecutionPool(extraMemoryNeeded: Long): Unit = {
if (extraMemoryNeeded > 0) {
//判断storage空闲空间和向execution借去的空间，取较大的那个。
//如果需要，storage会将先前向execution借的内存取消缓存进行返还。
val memoryReclaimableFromStorage =
math.max(storageMemoryPool.memoryFree, storageMemoryPool.poolSize - storageRegionSize)
//判断是否还有内存可以借出
if (memoryReclaimableFromStorage > 0) {
//如果剩余内存不够申请的，则将可用内存全部借出。
val spaceReclaimed = storageMemoryPool.shrinkPoolToFreeSpace(
math.min(extraMemoryNeeded, memoryReclaimableFromStorage))
onHeapExecutionMemoryPool.incrementPoolSize(spaceReclaimed)
}
}
}
storage向execution“借”内存
同样的，在storage需要内存的时候，execution也会将它的空闲内存借出去。因为实现复杂，所以execution空间不会被storage驱逐。这种机制带来的问题是，如果execution占据了storage大部分空间。这时候对一些数据的缓存可能会失败。
在下面的方法中可以清楚的看出，storage可使用的最大内存是maxMemory-execution已使用的内存数，意思就是说，即使execution占据了storage的空间也不会被收回，直到execution自己释放。
override def maxStorageMemory: Long = synchronized {
maxMemory - onHeapExecutionMemoryPool.memoryUsed
}
storage申请在acquireExecutionMemory方法中进行了实现。
override def acquireStorageMemory(
blockId: BlockId,
numBytes: Long,
evictedBlocks: mutable.Buffer[(BlockId, BlockStatus)]): Boolean = synchronized {
//如果需要空间大于Storage现有的最大内存空间，直接返回失败
if (numBytes > maxStorageMemory) {
return false
}
//如果需要空间比当前storage剩余空间多，则去借execution的空闲空间。
if (numBytes > storageMemoryPool.memoryFree) {
val memoryBorrowedFromExecution = Math.min(onHeapExecutionMemoryPool.memoryFree, numBytes)
onHeapExecutionMemoryPool.decrementPoolSize(memoryBorrowedFromExecution)
storageMemoryPool.incrementPoolSize(memoryBorrowedFromExecution)
}
storageMemoryPool.acquireMemory(blockId, numBytes, evictedBlocks)
更多实现细节可以去org.apache.spark.memory.UnifiedMemoryManager代码中查看。
带来的好处
除开极端内存使用环境（execution和storage空间都被占满）的情况下，新的内存管理机制能够减少shuffle过程中将数据spill到磁盘的次数或者提高数据的缓存比例，都能对程序运行速度带来一定的提升，同时也减少了调试参数的次数。
