apache-spark - 卡夫卡+ Spark 流: ClosedChannelException

Question

我正在尝试使用 Kafka 运行 Spark Streaming 作业。

我通过 Kafka 发送一个 csv 日志文件，以便它向我的 Spark Streaming 应用程序发布消息。

我在我的 Spark Streaming 应用程序中使用直接方法来实现这一点。

我的日志文件中的数据一开始很好地插入，但过了一会儿我在我的 Scala IDE 上看到以下错误消息。

环境:我在所有内核上本地运行 Spark。 Zookeeper，Kafka 也在我的系统上本地运行。

错误:

16/09/05 17:53:28 ERROR Executor: Exception in task 0.0 in stage 390.0 (TID 390)
java.nio.channels.ClosedChannelException
    at kafka.network.BlockingChannel.send(BlockingChannel.scala:100)
    at kafka.consumer.SimpleConsumer.liftedTree1$1(SimpleConsumer.scala:78)
    at kafka.consumer.SimpleConsumer.kafka$consumer$SimpleConsumer$$sendRequest(SimpleConsumer.scala:68)
    at kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$apply$mcV$sp$1.apply$mcV$sp(SimpleConsumer.scala:112)
    at kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$apply$mcV$sp$1.apply(SimpleConsumer.scala:112)
    at kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$apply$mcV$sp$1.apply(SimpleConsumer.scala:112)
    at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)
    at kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply$mcV$sp(SimpleConsumer.scala:111)
    at kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply(SimpleConsumer.scala:111)
    at kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply(SimpleConsumer.scala:111)
    at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33)

发生的事情是这个异常在中途抛出，有时控制台上会在这个错误消息之后出现一堆日志数据。

我有点困惑，因为这看起来像是一个网络错误 - “Closed Channel Exception”，但由于我在本地运行所有这些进程，我想知道是否其他原因可能真的是根本原因。

如果我能得到一些解决此问题的建议，那就太好了。

Answer 1

在 kafka producer.properties 配置(如 metadata.broker.list)中用你的机器 IP 替换 localhost。同样在/etc/hosts 文件中替换:

127.0.0.1 localhost localhost.localdomain

与

x.x.x.x localhost localhost.localdomain

其中 x.x.x.x 是您的机器 IP。看看是否有帮助。