Spark Shuffle 堆外内存溢出问题与解决(Shuffle通信原理)

Posted 南国故人

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Spark Shuffle 堆外内存溢出问题与解决(Shuffle通信原理)相关的知识,希望对你有一定的参考价值。

 

Spark Shuffle 堆外内存溢出问题与解决(Shuffle通信原理)

问题描述

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)

 

从失败信息可以看出,是堆外内存溢出问题,为什么会出现堆外内存溢出呢?

  Sparkshuffle部分使用了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)

 

 

当数据量大时能否直接写磁盘

MapReduceShuffle数据量大时,会把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需要被fetchblock全部封装成List<ManagedBuffer>,然后注册为一个Stream,然后把streamIdblockid的个数返回给客户端,最后返回给客户端的信息为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小于100Mspark.shuffle.max.block.size.inmemory)时,使用原有的方式Fetch数据,如果大于100M时,则使用新增的FetchStream协议,服务端在处理FetchStreamRequestFetchRequest的区别在于,FetchStreamRequest返回数据流,客户端根据返回的数据量写到本地临时文件,然后构造FileSegmentManagedBuffer给后续处理流程。

 

 

 

 

以上是关于Spark Shuffle 堆外内存溢出问题与解决(Shuffle通信原理)的主要内容,如果未能解决你的问题,请参考以下文章

Spark开发-Spark内存溢出原因以及解决方式

Spark面对OOM问题的解决方法及优化总结

如何优化 Apache Spark 应用程序中的 shuffle 溢出

spark 内存管理

Spark学习之路 SparkCore的调优之Spark内存模型

Spark面对OOM问题的解决方法及优化总结