Spark Streaming 作业在运行约 1 小时后被终止答案

【问题标题】：Spark Streaming job get killed after running for about 1 hourSpark Streaming 作业在运行约 1 小时后被终止
【发布时间】：2016-04-26 22:11:07
【问题描述】：

我有一个 spark 流作业，从 gnip 读取推文流并将其写入 Kafak。

Spark 和 kafka 在同一个集群上运行。

我的集群由 5 个节点组成。卡夫卡-b01 ...卡夫卡-b05

Spark master 正在 Kafak-b05 上运行。

这是我们提交 Spark 作业的方式

nohup sh $SPZRK_HOME/bin/spark-submit --total-executor-cores 5 --class com.test.java.gnipStreaming.GnipSparkStreamer --master spark://kafka-b05:7077 GnipStreamContainer。 jar powertrack kafka-b01,kafka-b02,kafka-b03,kafka-b04,kafka-b05 gnip_live_stream 2 &

大约 1 小时后，火花作业被终止

nohub 文件中的日志显示如下异常

org.apache.spark.storage.BlockFetchException: Failed to fetch block from 2 locations. Most recent failure cause: 
        at org.apache.spark.storage.BlockManager$$anonfun$doGetRemote$2.apply(BlockManager.scala:595) 
        at org.apache.spark.storage.BlockManager$$anonfun$doGetRemote$2.apply(BlockManager.scala:585) 
        at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59) 
        at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47) 
        at org.apache.spark.storage.BlockManager.doGetRemote(BlockManager.scala:585) 
        at org.apache.spark.storage.BlockManager.getRemote(BlockManager.scala:570) 
        at org.apache.spark.storage.BlockManager.get(BlockManager.scala:630) 
        at org.apache.spark.rdd.BlockRDD.compute(BlockRDD.scala:48) 
        at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:306) 
        at org.apache.spark.rdd.RDD.iterator(RDD.scala:270) 
        at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66) 
        at org.apache.spark.scheduler.Task.run(Task.scala:89) 
        at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214) 
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) 
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) 
        at java.lang.Thread.run(Thread.java:745) 
Caused by: io.netty.channel.ChannelException: Unable to create Channel from class class io.netty.channel.socket.nio.NioSocketChannel 
        at io.netty.bootstrap.AbstractBootstrap$BootstrapChannelFactory.newChannel(AbstractBootstrap.java:455) 
        at io.netty.bootstrap.AbstractBootstrap.initAndRegister(AbstractBootstrap.java:306) 
        at io.netty.bootstrap.Bootstrap.doConnect(Bootstrap.java:134) 
        at io.netty.bootstrap.Bootstrap.connect(Bootstrap.java:116) 
        at org.apache.spark.network.client.TransportClientFactory.createClient(TransportClientFactory.java:211) 
        at org.apache.spark.network.client.TransportClientFactory.createClient(TransportClientFactory.java:167) 
        at org.apache.spark.network.netty.NettyBlockTransferService$$anon$1.createAndStart(NettyBlockTransferService.scala:90) 
        at org.apache.spark.network.shuffle.RetryingBlockFetcher.fetchAllOutstanding(RetryingBlockFetcher.java:140) 
        at org.apache.spark.network.shuffle.RetryingBlockFetcher.start(RetryingBlockFetcher.java:120) 
        at org.apache.spark.network.netty.NettyBlockTransferService.fetchBlocks(NettyBlockTransferService.scala:99) 
        at org.apache.spark.network.BlockTransferService.fetchBlockSync(BlockTransferService.scala:89) 
        at org.apache.spark.storage.BlockManager$$anonfun$doGetRemote$2.apply(BlockManager.scala:588) 
        ... 15 more 
Caused by: io.netty.channel.ChannelException: Failed to open a socket. 
        at io.netty.channel.socket.nio.NioSocketChannel.newSocket(NioSocketChannel.java:62) 
        at io.netty.channel.socket.nio.NioSocketChannel.<init>(NioSocketChannel.java:72) 
        at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method) 
        at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62) 
        at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45) 
        at java.lang.reflect.Constructor.newInstance(Constructor.java:423) 
        at java.lang.Class.newInstance(Class.java:442) 
        at io.netty.bootstrap.AbstractBootstrap$BootstrapChannelFactory.newChannel(AbstractBootstrap.java:453) 
        ... 26 more 
Caused by: java.net.SocketException: Too many open files 
        at sun.nio.ch.Net.socket0(Native Method) 
        at sun.nio.ch.Net.socket(Net.java:411) 
        at sun.nio.ch.Net.socket(Net.java:404) 
        at sun.nio.ch.SocketChannelImpl.<init>(SocketChannelImpl.java:105) 
        at sun.nio.ch.SelectorProviderImpl.openSocketChannel(SelectorProviderImpl.java:60) 
        at io.netty.channel.socket.nio.NioSocketChannel.newSocket(NioSocketChannel.java:60) 
        ... 33 more

我已将打开文件的最大数量增加到 3275782（旧数字几乎是这个数字的一半），但我仍然面临同样的问题。

当我从 spark Web 界面检查工作人员的 stderr 日志时，我发现了另一个异常。

java.nio.channels.ClosedChannelException 
        at kafka.network.BlockingChannel.send(BlockingChannel.scala:110) 
        at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:75) 
        at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:74) 
        at kafka.producer.SyncProducer.send(SyncProducer.scala:119) 
        at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:59) 
        at kafka.producer.BrokerPartitionInfo.updateInfo(BrokerPartitionInfo.scala:82) 
        at kafka.producer.BrokerPartitionInfo.getBrokerPartitionInfo(BrokerPartitionInfo.scala:49) 
        at kafka.producer.async.DefaultEventHandler.kafka$producer$async$DefaultEventHandler$$getPartitionListForTopic(DefaultEventHandler.scala:188) 
        at kafka.producer.async.DefaultEventHandler$$anonfun$partitionAndCollate$1.apply(DefaultEventHandler.scala:152) 
        at kafka.producer.async.DefaultEventHandler$$anonfun$partitionAndCollate$1.apply(DefaultEventHandler.scala:151) 
        at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59) 
        at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47) 
        at kafka.producer.async.DefaultEventHandler.partitionAndCollate(DefaultEventHandler.scala:151) 
        at kafka.producer.async.DefaultEventHandler.dispatchSerializedData(DefaultEventHandler.scala:96) 
        at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:73) 
        at kafka.producer.Producer.send(Producer.scala:77) 
        at kafka.javaapi.producer.Producer.send(Producer.scala:33) 
        at com.test.java.gnipStreaming.GnipSparkStreamer$1$1.call(GnipSparkStreamer.java:59) 
        at com.test.java.gnipStreaming.GnipSparkStreamer$1$1.call(GnipSparkStreamer.java:51) 
        at org.apache.spark.api.java.JavaRDDLike$$anonfun$foreachPartition$1.apply(JavaRDDLike.scala:225) 
        at org.apache.spark.api.java.JavaRDDLike$$anonfun$foreachPartition$1.apply(JavaRDDLike.scala:225) 
        at org.apache.spark.rdd.RDD$$anonfun$foreachPartition$1$$anonfun$apply$33.apply(RDD.scala:920) 
        at org.apache.spark.rdd.RDD$$anonfun$foreachPartition$1$$anonfun$apply$33.apply(RDD.scala:920) 
        at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1858) 
        at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1858) 
        at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66) 
        at org.apache.spark.scheduler.Task.run(Task.scala:89) 
        at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214) 
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) 
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) 
        at java.lang.Thread.run(Thread.java:745)

第二个例外（看起来）与卡夫卡而不是火花有关。

你认为问题是什么？

编辑

基于 Yuval Itzchakov 的评论这是流媒体的代码

主类http://pastebin.com/EcbnQQ3a

客户接收类http://pastebin.com/3UFPktKR

【问题讨论】：

您需要向我们展示写入 Kafka 的代码。并发连接数不应超过您在 Kafka 中设置的每个主题的分区数。增加文件句柄的数量只是一个补丁，你很可能在代码中做错了。
Thnx Yuval，我为代码添加了两个链接。顺便说一句，当我将相同的作业提交到本地 spark 安装时，它可以正常工作，没有任何问题。我怀疑同一个 Spark 集群上存在 Kafka 是这个问题的根本原因。你怎么看？

标签： apache-spark apache-kafka spark-streaming

【解决方案1】：

问题是您在 DStream.foreachPartition 的迭代中实例化了 Producer 的新实例。如果您有一个数据密集型流，这可能会导致分配大量生产者并尝试连接到 Kafka。

我要确保的第一件事是，一旦您使用finally 块发送数据并调用producer.close，您就正确地关闭了流：

public Void call(JavaRDD<String> rdd) throws Exception {
    rdd.foreachPartition(new VoidFunction<Iterator<String>>() {

        @Override
        public void call(Iterator<String> itr) throws Exception {
                            try
                            {
               Producer<String, String> producer = getProducer(hosts);
               while(itr.hasNext()) {
                 try {
                    KeyedMessage<String, String> message = 
                        new KeyedMessage<String, String>(topic, itr.next());
                    producer.send(message);
                   } catch (Exception e) {
                    e.printStackTrace();
                   }
               } finally {
                                   producer.close()
                               }
        }
    });
    return null;
}

如果这仍然不起作用并且您看到太多连接，我会为 Kafka 生产者创建一个对象池，您可以按需进行池化。这样，您就可以明确控制正在使用的可用生产者的数量以及您打开的套接字数量。

【讨论】：

我添加了 producer.close() 行，并且工作在 14 小时前工作顺利。谢谢尤瓦尔。
Yuval，关闭生产者使作业一直运行，但现在工人的标准错误充满了“kafka.producer.ProducerClosedException：生产者已经关闭”异常。你对此有什么想法吗？
你是否在 finally 块中关闭它们？
是的，正如你所说。我做了两个改变，解决了我所有的问题。第一个是提交作业时使用--executor-memory 2G。另一种是在while循环外关闭生产者。作业到现在已经运行了 4 个小时，stderr 中没有任何异常。
@Fanooos 哎呀，我忘了我们在迭代器中。当然，这是解决方法 :) 在循环外添加 try-finally。