Spark Shuffle 堆外内存溢出问题与解决（Shuffle通信原理）
http://xiguada.org/spark-shuffle-direct-buffer-oom/
问题描述
Spark-1.6.0已经在一月份release，为了验证一下它的性能，我使用了一些大的SQL验证其性能，其中部分SQL出现了Shuffle失败问题，详细的堆栈信息如下所示：
16/02/17 15:36:36 WARN server.TransportChannelHandler: Exception in connection from /10.196.134.220:7337
java.lang.OutOfMemoryError: Direct buffer memory
at java.nio.Bits.reserveMemory(Bits.java:658)
at java.nio.DirectByteBuffer.<init>(DirectByteBuffer.java:123)
at java.nio.ByteBuffer.allocateDirect(ByteBuffer.java:306)
at io.netty.buffer.PoolArena$DirectArena.newChunk(PoolArena.java:645)
at io.netty.buffer.PoolArena.allocateNormal(PoolArena.java:228)
at io.netty.buffer.PoolArena.allocate(PoolArena.java:212)
at io.netty.buffer.PoolArena.allocate(PoolArena.java:132)
at io.netty.buffer.PooledByteBufAllocator.newDirectBuffer(PooledByteBufAllocator.java:271)
at io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:155)
at io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:146)
at io.netty.buffer.AbstractByteBufAllocator.ioBuffer(AbstractByteBufAllocator.java:107)
at io.netty.channel.AdaptiveRecvByteBufAllocator$HandleImpl.allocate(AdaptiveRecvByteBufAllocator.java:104)
at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:117)
at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:511)
at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:468)
at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:382)
at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:354)
at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:111)
at java.lang.Thread.run(Thread.java:744)
从失败信息可以看出，是堆外内存溢出问题，为什么会出现堆外内存溢出呢？
Spark的shuffle部分使用了netty框架进行网络传输，但netty会申请堆外内存缓存（PooledByteBufAllocator ，AbstractByteBufAllocator）；Shuffle时，每个Reduce都需要获取每个map对应的输出，当一个reduce需要获取的一个map数据比较大（比如1G），这时候就会申请一个1G的堆外内存，而堆外内存是有限制的，这时候就出现了堆外内存溢出。
Shuffle不使用堆外内存
为Executor增加配置-Dio.netty.noUnsafe=true，就可以让shuffle不使用堆外内存，但相同的作业还是出现了OOM，这种方式没办法解决问题。
java.lang.OutOfMemoryError: Java heap space
at io.netty.buffer.PoolArena$HeapArena.newUnpooledChunk(PoolArena.java:607)
at io.netty.buffer.PoolArena.allocateHuge(PoolArena.java:237)
at io.netty.buffer.PoolArena.allocate(PoolArena.java:215)
at io.netty.buffer.PoolArena.allocate(PoolArena.java:132)
at io.netty.buffer.PooledByteBufAllocator.newHeapBuffer(PooledByteBufAllocator.java:256)
at io.netty.buffer.AbstractByteBufAllocator.heapBuffer(AbstractByteBufAllocator.java:136)
at io.netty.buffer.AbstractByteBufAllocator.heapBuffer(AbstractByteBufAllocator.java:127)
at io.netty.buffer.CompositeByteBuf.allocBuffer(CompositeByteBuf.java:1347)
at io.netty.buffer.CompositeByteBuf.consolidateIfNeeded(CompositeByteBuf.java:276)
at io.netty.buffer.CompositeByteBuf.addComponent(CompositeByteBuf.java:116)
at org.apache.spark.network.util.TransportFrameDecoder.decodeNext(TransportFrameDecoder.java:148)
at org.apache.spark.network.util.TransportFrameDecoder.channelRead(TransportFrameDecoder.java:82)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:308)
at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:294)
at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:846)
at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:131)
at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:511)
at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:468)
at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:382)
当数据量大时能否直接写磁盘
MapReduce中Shuffle数据量大时，会把Shuffle数据写到磁盘。
Spark Shuffle通信机制
上图显示了Shuffle的通信原理。
服务端会启动Shuffle_Service。
（1）客户端代码调用堆栈
BlockStoreShuffleReader.read
ShuffleBlockFetcherIterator.sendRequest
ExternalShuffleClient.fetchBlocks
OneForOneBlockFetcher.start
TransportClient.sendRpc
发送RpcRequest(OpenBlocks)信息
（2）服务端代码调用堆栈
TransportRequestHandler.processRpcRequest
ExternalShuffleBlockHandler.receive
ExternalShuffleBlockHandler.handleMessage
ExternalShuffleBlockResolver.getBlockData(shuffle_ShuffleId_MapId_ReduceId)
ExternalShuffleBlockResolver.getSortBasedShuffleBlockData
FileSegmentManagedBuffer
handleMessage会把所需的appid的一个executor需要被fetch的block全部封装成List<ManagedBuffer>，然后注册为一个Stream，然后把streamId和blockid的个数返回给客户端，最后返回给客户端的信息为RpcResponse(StreamHandle(streamId, msg.blockIds.length))。
（3）客户端
客户端接收到RpcResponse后，会为每个blockid调用：
TransportClient.fetchChunk
Send ChunkFetchRequest(StreamChunkId(streamId, chunkIndex))
（4）服务端
TransportRequestHandler.processFetchRequest
OneForOneStreamManager.getChunk
返回respond(new ChunkFetchSuccess(req.streamChunkId, buf))给客户端，buf就是某一个blockid的FileSegmentManagedBuffer。
（5）客户端
OneForOneBlockFetcher.ChunkCallback.onSuccess
listener.onBlockFetchSuccess(blockIds[chunkIndex], buffer)
ShuffleBlockFetcherIterator.sendRequest.BlockFetchingListener.onBlockFetchSuccess
results.put(new SuccessFetchResult(BlockId(blockId), address, sizeMap(blockId), buf))
客户端的另外一个线程
ShuffleBlockFetcherIterator.next
(result.blockId, new BufferReleasingInputStream(buf.createInputStream(), this))
Download文件的通信原理
另外还有一个stream通信协议，客户端首先需要构造StreamRequest请求，StreamRequest中包含待下载文件的URL。
（1）客户端调用堆栈
Executor.updateDependencies...
org.apache.spark.util.Utils.fetchFile
org.apache.spark.util.Utils.doFetchFile
NettyRpcEnv.openChannel
TransportClient.stream
Send StreamRequest(streamId) streamId为文件的目录。
（2）服务端处理流程
TransportRequestHandler.handle
TransportRequestHandler.processStreamRequest
OneForOneStreamManager.openStream
返回new StreamResponse(req.streamId, buf.size(), buf)
（3）客户端处理流程
TransportResponseHandler.handle
TransportFrameDecoder.channelRead
TransportFrameDecoder.feedInterceptor
StreamInterceptor.handle
callback.onData即NettyRpcEnv.FileDownloadCallback.onData
然后返回client.stream(parsedUri.getPath(), callback)给Utils.doFetchFile，最后org.apache.spark.util.Utils.downloadFile
问题分析：
当前spark shuffle时使用Fetch协议，由于使用堆外内存存储Fetch的数据，当Fetch某个map的数据特别大时，容易出现堆外内存的OOM。而申请内存部分在Netty自带的代码中，我们无法修改。
另外一方面，Stream是下载文件的协议，需要提供文件的URL，而Shuffle只会获取文件中的一段数据，并且也不知道URL，因此不能直接使用Stream接口。
解决方案：
新增一个FetchStream通信协议，在OneForOneBlockFetcher中，如果一个block小于100M（spark.shuffle.max.block.size.inmemory）时，使用原有的方式Fetch数据，如果大于100M时，则使用新增的FetchStream协议，服务端在处理FetchStreamRequest和FetchRequest的区别在于，FetchStreamRequest返回数据流，客户端根据返回的数据量写到本地临时文件，然后构造FileSegmentManagedBuffer给后续处理流程。