Spark Sort Based Shuffle 内存使用情况

Spark Shuffle Write 调用链：

ShuffleMapTask
---> SortShuffleManager
---> SortShuffleWriter
---> ExternalSorter

在ExternalSorter发生了分区数据写内存、聚合、Spill、Merge、写文件操作。

这里的内存结构为：

PartitionedAppendOnlyMap[K, C]

本质上是定长数组：

private var data = new Array[AnyRef](2 * capacity)

当这个内存数据写不下的时候，会发生spill，内存数据写入临时文件，写文件的时候需要一个Buffer：

spark.shuffle.file.buffer=32k

读写数据需要的序列化/反序列化存储空间：

spark.shuffle.spill.batchSize=10000

如果某一时刻Executor运行mapTask任务的Core数目为 C， 则shuffle占用的内存空间为：

C * 32k + C * 10000个Record + C * PartitionedAppendOnlyMap.size

其中：

C * PartitionedAppendOnlyMap.size < ExecutorHeapMemeory * 0.2 * 0.8

这部分是由

ShuffleMemoryManager

来管理的(具体细节还要在看代码)。

spark.shuffle.memoryFraction=0.2
spark.shuffle.safetyFraction=0.8

【Note】

在判断是否需要spill到磁盘时，常规实现是：每写一条数据到内存就进行一次判断，假设判断

PartitionedAppendOnlyMap

占用多少空间需要1ms，那么写1000万条数据时，这个判断消耗的总时间就难以接受。

实际中 通过 estimateSize 函数来进行采样判断，并不是每次都进行判断。采样的周期不是固定的，而是指数增长的，比如2次操作后采样判断，那么接下来的采样判断会在2*2次操作后发生，依次类推，即越往后操作的次数越多才会发生一次采样判断。

这样的一个问题就是： 如果你的shuffle内存开的很大，那PartitionedAppendOnlyMap可能要经过几十万次更新之后之后才会进行一次采样，才能计算出新的大小，这个时候几十万次更新带来的新的内存压力，可能已经让你的GC不堪重负了。所以 shuffle内存不能盲目的开的很大，适中即可



Shuffle Read的内存使用基本和上面描述的差不多。