Spark streaming kafka找不到集合的领导者偏移

我使用spark spark'org.apache.spark：spark-streaming_2.10：1.6.1'和'org.apache.spark：spark-streaming-kafka_2.10 ：1.6.1'连接到一个kafka经纪人版本0.10.0.1。当我尝试此代码：Spark streaming kafka找不到集合的领导者偏移

def messages = KafkaUtils.createDirectStream(jssc, 
      String.class, 
      String.class, 
      StringDecoder.class, 
      StringDecoder.class, 
      kafkaParams, 
      topicsSet)

我已经收到此异常：

INFO consumer.SimpleConsumer: Reconnect due to socket error: java.nio.channels.ClosedChannelException 
Exception in thread "main" org.apache.spark.SparkException: java.nio.channels.ClosedChannelException 
org.apache.spark.SparkException: Couldn't find leader offsets for Set([stream,0]) 
    at org.apache.spark.streaming.kafka.KafkaCluster$$anonfun$checkErrors$1.apply(KafkaCluster.scala:366) 
    at org.apache.spark.streaming.kafka.KafkaCluster$$anonfun$checkErrors$1.apply(KafkaCluster.scala:366) 
    at scala.util.Either.fold(Either.scala:97) 
    at org.apache.spark.streaming.kafka.KafkaCluster$.checkErrors(KafkaCluster.scala:365) 
    at org.apache.spark.streaming.kafka.KafkaUtils$.getFromOffsets(KafkaUtils.scala:222) 
    at org.apache.spark.streaming.kafka.KafkaUtils$.createDirectStream(KafkaUtils.scala:484) 
    at org.apache.spark.streaming.kafka.KafkaUtils$.createDirectStream(KafkaUtils.scala:607) 
    at org.apache.spark.streaming.kafka.KafkaUtils.createDirectStream(KafkaUtils.scala) 
    at org.apache.spark.streaming.kafka.KafkaUtils$createDirectStream.call(Unknown Source) 
    at org.codehaus.groovy.runtime.callsite.CallSiteArray.defaultCall(CallSiteArray.java:45) 
    at org.codehaus.groovy.runtime.callsite.AbstractCallSite.call(AbstractCallSite.java:108) 
    at com.privowny.classification.jobs.StreamingClassification.main(StreamingClassification.groovy:48) 
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) 
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) 
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) 
    at java.lang.reflect.Method.invoke(Method.java:483) 
    at org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:731) 
    at org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:181) 
    at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:206) 
    at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:121) 
    at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)

我尝试搜索这个网站的一些答案，但似乎没有答案，你可以给我一些建议做什么？主题stream不是空的。

来源

2016-09-01 innovatism

这通常是ZooKeeper问题的一个信号。重置ZooKeeper并重试。 –

可能是什么问题？我刚刚启动了服务器，就像在快速入门文档中一样！ – innovatism

我遇到了Kafka和ZooKeeper之间存在同步问题的问题。重置他们都解决了它。 –

我从经验中知道，如果Spark驱动程序无法使用经纪人公布的主机名（advertised.host.name在server.properties）到达kafka经纪人，那么可能导致此错误消息的一件事是。即使spark配置识别使用不同地址的kafka代理，情况也是如此。所有代理公布的主机名都必须可以从Spark驱动程序中获得。

这发生在我身上，因为集群运行在单独的AWS账户中，经纪人使用内部DNS记录识别自己，这些记录必须复制到其他AWS账户。在这之前，我得到了这个错误信息，因为即使我们在spark配置中使用经纪人的私有IP地址，Spark驱动程序也无法联系到经纪人询问他们最新的抵消额。

希望能帮助别人。

来源

2016-10-28 12:26:39

我也遇到过这个问题。所以你必须改变卡夫卡的一些配置。

转到您的Kafka配置并配置listeners;

在格式套接字服务器设置部分：

listeners=PLAINTEXT://[hostname or IP]:[port]

例如：

listeners=PLAINTEXT://192.168.1.24:9092

来源

2016-12-07 09:50:22

我跑卡夫卡从HDP，所以默认端口是6667，而不是9092，当我将bootstrap.servers的端口切换为<hostname>:6667，问题得到解决。

来源

2018-01-27 23:45:05

Spark streaming kafka找不到集合的领导者偏移

回答

相关问题