有未完成的请求时，外部 Shuffle 服务连接空闲超过 120 秒答案

【问题标题】：External Shuffle service connection idle for more than 120seconds while there are outstanding requests有未完成的请求时，外部 Shuffle 服务连接空闲超过 120 秒
【发布时间】：2018-12-10 08:38:46
【问题描述】：

我正在纱线上运行火花作业。该作业在亚马逊 EMR 上正常运行。（1 个主控和 2 个从属，带有 m4.xlarge）

我使用 aws ec2 机器使用 HDP 2.6 分发设置了类似的基础设施。但是火花作业卡在一个特定阶段，一段时间后我在容器日志中收到以下错误。主要错误似乎是 shuffle 服务处于空闲状态。

18/06/25 07:15:31 信息 spark.MapOutputTrackerWorker：进行提取；跟踪器端点 = NettyRpcEndpointRef(spark://MapOutputTracker@10.210.150.150:44343) 18/06/25 07:15:31 INFO spark.MapOutputTrackerWorker：没有 shuffle 9 的地图输出，正在获取它们 18/06/25 07:15:31 INFO spark.MapOutputTrackerWorker：没有 shuffle 9 的地图输出，正在获取它们 18/06/25 07:15:31 INFO spark.MapOutputTrackerWorker：得到输出位置 18/06/25 07:15:31 INFO storage.ShuffleBlockFetcherIterator：从 1000 个块中获取 5 个非空块 18/06/25 07:15:31 INFO storage.ShuffleBlockFetcherIterator：在 0 毫秒内开始 1 次远程提取 18/06/25 07:15:31 INFO storage.ShuffleBlockFetcherIterator：从 1000 个块中获取 5 个非空块 18/06/25 07:15:31 INFO storage.ShuffleBlockFetcherIterator：在 0 毫秒内开始 0 次远程获取 18/06/25 07:15:31 INFO storage.ShuffleBlockFetcherIterator：从 1000 个块中获取 5 个非空块 18/06/25 07:15:31 INFO storage.ShuffleBlockFetcherIterator：在 0 毫秒内开始 1 次远程提取 18/06/25 07:15:31 INFO storage.ShuffleBlockFetcherIterator：从 1000 个块中获取 5 个非空块 18/06/25 07:15:31 INFO storage.ShuffleBlockFetcherIterator：在 1 毫秒内开始 1 次远程提取 18/06/25 07:15:31 INFO codegen.CodeGenerator：在 4.822611 毫秒内生成的代码 18/06/25 07:15:31 INFO codegen.CodeGenerator：在 8.430244 毫秒内生成的代码 18/06/25 07:17:31 ERROR server.TransportChannelHandler: Connection to ip-10-210-150-180.********/10.210.150.180:7447 已经安静了 120000 毫秒，而有未完成的请求。假设连接已死；如果这是错误的，请调整 spark.network.timeout。 18/06/25 07:17:31 错误 client.TransportResponseHandler：当来自 ip-10-210-150-180.********/10.210.150.180:7447 的连接关闭时，仍有 307 个请求未完成 18/06/25 07:17:31 INFO shuffle.RetryingBlockFetcher：在 5000 毫秒后重试 197 个未完成块的提取（1/3） 18/06/25 07:17:31 错误 shuffle.OneForOneBlockFetcher：获取起始块时失败 java.io.IOException：来自 ip-10-210-150-180.********/10.210.150.180:7447 的连接已关闭在 org.apache.spark.network.client.TransportResponseHandler.channelInactive(TransportResponseHandler.java:146) 在 org.apache.spark.network.server.TransportChannelHandler.channelInactive(TransportChannelHandler.java:108) 在 io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:241) 在 io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:227) 在 io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:220) 在 io.netty.channel.ChannelInboundHandlerAdapter.channelInactive(ChannelInboundHandlerAdapter.java:75) 在 io.netty.handler.timeout.IdleStateHandler.channelInactive(IdleStateHandler.java:278) 在 io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:241) 在 io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:227) 在 io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:220) 在 io.netty.channel.ChannelInboundHandlerAdapter.channelInactive(ChannelInboundHandlerAdapter.java:75) 在 io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:241) 在 io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:227) 在 io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:220) 在 io.netty.channel.ChannelInboundHandlerAdapter.channelInactive(ChannelInboundHandlerAdapter.java:75) 在 org.apache.spark.network.util.TransportFrameDecoder.channelInactive（TransportFrameDecoder.java:182）在 io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:241) 在 io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:227) 在 io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:220) 在 io.netty.channel.DefaultChannelPipeline$HeadContext.channelInactive(DefaultChannelPipeline.java:1289) 在 io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:241) 在 io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:227) 在 io.netty.channel.DefaultChannelPipeline.fireChannelInactive（DefaultChannelPipeline.java:893）在 io.netty.channel.AbstractChannel$AbstractUnsafe$7.run(AbstractChannel.java:691) 在 io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks（SingleThreadEventExecutor.java:399）在 io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:446) 在 io.netty.util.concurrent.SingleThreadEventExecutor$2.run（SingleThreadEventExecutor.java:131）在 io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:144) 在 java.lang.Thread.run(Thread.java:748) 18/06/25 07:17:31 INFO shuffle.RetryingBlockFetcher：在 5000 毫秒后重试 166 个未完成块的提取（1/3） 18/06/25 07:17:31 错误 shuffle.OneForOneBlockFetcher：获取起始块时失败 java.io.IOException：来自 ip-10-210-150-180.********/10.210.150.180:7447 的连接已关闭在 org.apache.spark.network.client.TransportResponseHandler.channelInactive(TransportResponseHandler.java:146) 在 org.apache.spark.network.server.TransportChannelHandler.channelInactive(TransportChannelHandler.java:108) 在 io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:241) 在 io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:227) 在 io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:220) 在 io.netty.channel.ChannelInboundHandlerAdapter.channelInactive(ChannelInboundHandlerAdapter.java:75) 在 io.netty.handler.timeout.IdleStateHandler.channelInactive(IdleStateHandler.java:278) 在 io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:241) 在 io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:227) 在 io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:220) 在 io.netty.channel.ChannelInboundHandlerAdapter.channelInactive(ChannelInboundHandlerAdapter.java:75) 在 io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:241) 在 io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:227) 在 io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:220) 在 io.netty.channel.ChannelInboundHandlerAdapter.channelInactive(ChannelInboundHandlerAdapter.java:75) 在 org.apache.spark.network.util.TransportFrameDecoder.channelInactive（TransportFrameDecoder.java:182）在 io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:241) 在 io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:227) 在 io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:220) 在 io.netty.channel.DefaultChannelPipeline$HeadContext.channelInactive(DefaultChannelPipeline.java:1289) 在 io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:241) 在 io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:227) 在 io.netty.channel.DefaultChannelPipeline.fireChannelInactive（DefaultChannelPipeline.java:893）在 io.netty.channel.AbstractChannel$AbstractUnsafe$7.run(AbstractChannel.java:691) 在 io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks（SingleThreadEventExecutor.java:399）在 io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:446) 在 io.netty.util.concurrent.SingleThreadEventExecutor$2.run（SingleThreadEventExecutor.java:131）在 io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:144) 在 java.lang.Thread.run(Thread.java:748)

我目前正在使用以下 spark-defaults 配置在 yarn 集群上运行 spark

spark.eventLog.dir=hdfs:///user/spark/applicationHistory
spark.eventLog.enabled=true
spark.yarn.historyServer.address=ppv-qa12-tenant8-spark-cluster-master.periscope-solutions.local:18080
spark.shuffle.service.enabled=true
spark.dynamicAllocation.enabled=true
spark.driver.extraLibraryPath=/usr/hdp/current/hadoop-client/lib/native:/usr/hdp/current/hadoop-client/lib/native/Linux-amd64-64
spark.executor.extraLibraryPath=/usr/hdp/current/hadoop-client/lib/native:/usr/hdp/current/hadoop-client/lib/native/Linux-amd64-64
spark.driver.maxResultSize=0
spark.driver.extraJavaOptions=-XX:+UseConcMarkSweepGC -XX:CMSInitiatingOccupancyFraction=70 -XX:MaxHeapFreeRatio=70 -XX:+CMSClassUnloadingEnabled -XX:OnOutOfMemoryError='kill -9 %p'
spark.executor.extraJavaOptions=-verbose:gc -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:+UseConcMarkSweepGC -XX:CMSInitiatingOccupancyFraction=70 -XX:MaxHeapFreeRatio=70 -XX:+CMSClassUnloadingEnabled -XX:OnOutOfMemoryError='kill -9 %p'
spark.executor.memory=5g
spark.driver.memory=1g
spark.executor.cores=4

我在从机节点管理器的 yarn-site.xml 中有以下设置

<configuration>
  <property>
    <name>yarn.application.classpath</name>
    <value>/usr/hdp/current/spark2-client/aux/*,/etc/hadoop/conf,/usr/hdp/current/hadoop-client/*,/usr/hdp/current/hadoop-client/lib/*,/usr/hdp/current/hadoop-hdfs-client/*,/usr/hdp/current/hadoop-hdfs-client/lib/*,/usr/hdp/current/hadoop-yarn-client/*,/usr/hdp/current/hadoop-yarn-client/lib/*</value>
  </property>
  <property>
    <name>yarn.nodemanager.aux-services</name>
    <value>spark2_shuffle</value>
  </property>
  <property>
    <name>yarn.nodemanager.aux-services.mapreduce_shuffle.class</name>
    <value>org.apache.hadoop.mapred.ShuffleHandler</value>
  </property>
  <property>
    <name>yarn.nodemanager.aux-services.spark2_shuffle.class</name>
    <value>org.apache.spark.network.yarn.YarnShuffleService</value>
  </property>
  <property>
    <name>yarn.nodemanager.container-manager.thread-count</name>
    <value>64</value>
  </property>
  <property>
    <name>yarn.nodemanager.localizer.client.thread-count</name>
    <value>20</value>
  </property>
  <property>
    <name>yarn.nodemanager.vmem-pmem-ratio</name>
    <value>5</value>
  </property>
  <property>
    <name>yarn.resourcemanager.hostname</name>
    <value>************</value>
  </property>
  <property>
    <name>yarn.resourcemanager.resource-tracker.client.thread-count</name>
    <value>64</value>
  </property>
  <property>
    <name>yarn.resourcemanager.scheduler.client.thread-count</name>
    <value>64</value>
  </property>
  <property>
    <name>yarn.scheduler.increment-allocation-mb</name>
    <value>32</value>
  </property>
  <property>
    <name>yarn.scheduler.increment-allocation-vcores</name>
    <value>1</value>
  </property>
  <property>
    <name>yarn.scheduler.maximum-allocation-vcores</name>
    <value>128</value>
  </property>
  <property>
    <name>yarn.scheduler.minimum-allocation-mb</name>
    <value>32</value>
  </property>
  <property>
    <name>yarn.timeline-service.enabled</name>
    <value>true</value>
  </property>
  <property>
  <name>yarn.nodemanager.resource.cpu-vcores</name>
    <value>8</value>
  </property>
  <property>
  <name>yarn.nodemanager.resource.memory-mb</name>
    <value>11520</value>
  </property>
  <property>
  <name>yarn.scheduler.maximum-allocation-mb</name>
    <value>11520</value>
  </property>
  <property>
  <name>yarn.nodemanager.hostname</name>
    <value>*************</value>
  </property>
</configuration>

编辑：通过一些网络调试，我发现容器创建的用于连接随机播放服务的临时端口正在主动拒绝连接。（telnet 立即抛出错误）

【问题讨论】：

能否请您确保洗牌服务是否在所有工人中正常运行？有时 shuffle 服务会静默停止并最终导致这些错误
没有在所有机器上运行随机播放服务。目前它作为yarn的辅助服务运行。

标签： amazon-web-services apache-spark hadoop-yarn

【解决方案1】：

在查看内核和系统活动日志时，我们在/var/log/messages 中发现了以下问题

xen_netfront: xennet: skb 乘坐火箭：19 个插槽

这意味着我们的 aws ec2 机器出现网络丢包。

数据传输 b/n 容器和 shuffle 服务通过 RPC 调用（ChunkFetchRequest、ChunkFetchSuccess 和 ChunkFetchFailure）发生，这些 RPC 调用被网络抑制。

有关此日志的更多信息可以在以下线程中找到。

http://www.brendangregg.com/blog/2014-09-11/perf-kernel-line-tracing.html

日志消息意味着我们超出了可以放入驱动程序环形缓冲区队列（16）的数据包的最大缓冲区大小，并且那些 SKB 丢失了

Scatter-gather 收集多个响应并将它们作为单个响应发送，这反过来又导致 SKB 大小的增加。

所以我们使用以下命令关闭了分散收集。

sudo ethtool -K eth0 sg off

在此之后没有更多的数据包丢失。

性能也与我们过去在 EMR 中的性能相似。

【讨论】：