我正在尝试设置Spark Streaming以从Kafka队列获取消息。我收到以下错误:

py4j.protocol.Py4JJavaError: An error occurred while calling o30.createDirectStream.
: org.apache.spark.SparkException: java.nio.channels.ClosedChannelException
org.apache.spark.SparkException: Couldn't find leader offsets for Set([test-topic,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)


这是我正在执行的代码(pyspark):

from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils

directKafkaStream = KafkaUtils.createDirectStream(ssc, ["test-topic"], {"metadata.broker.list": "host.domain:9092"})

ssc.start()
ssc.awaitTermination()


有几个类似的帖子,但有相同的错误。在所有情况下,原因都是空的kafka主题。我的“测试主题”中有消息。我可以带他们出去

kafka-console-consumer --zookeeper host.domain:2181 --topic test-topic --from-beginning --max-messages 100


有谁知道可能是什么问题?

我正在使用:


Spark 1.5.2(Apache)
Kafka 0.8.2.0 + kafka1.3.0(CDH 5.4.7)

最佳答案

您需要检查两件事:


检查此主题和分区是否存在,在您的情况下,主题为test-topic且分区为0。
根据您的代码,您正在尝试从偏移量0开始消耗消息,并且可能从偏移量0开始不可用消息,请检查最早的偏移量,然后尝试从偏移量开始消耗。


以下是检查最早的偏移量的命令:

sh kafka/bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list "your broker list" --topic "topic name" --time -1

关于apache-spark - Spark Streaming + Kafka:SparkException:找不到Set的引线偏移量,我们在Stack Overflow上找到一个类似的问题:https://stackoverflow.com/questions/34288449/

10-13 00:30