kafka.cluster.BrokerEndPoint无法转换为kafka.cluster.Broker [英] kafka.cluster.BrokerEndPoint cannot be cast to kafka.cluster.Broker

问题描述

我正在尝试将数据从kafka传输到spark

I'm trying to stream data from kafka to spark

我正在使用带有kafka 0.9.0.1和scala 2.11.8的spark 1.6.2

I'm using spark 1.6.2 with kafka 0.9.0.1 and scala 2.11.8

当我使用基于接收者的方法时，一切正常(KafkaUtils.createStream())但是当我尝试没有这种接收者的直接方法

everything works fine when I use the receiver-based approach(KafkaUtils.createStream()) but when I try the direct approach with no receivers like this

val kafkaStreams = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
  ssc,
  Map("group.id" -> "blah",
    "auto.offset.reset" -> "smallest",
    "metadata.broker.list" -> "127.0.0.1:9092",
    "bootstrap.servers"-> "127.0.0.1:9092"),
  Set("tweets")
  )

我收到此错误

Exception in thread "main" java.lang.ClassCastException: kafka.cluster.BrokerEndPoint cannot be cast to kafka.cluster.Broker
at org.apache.spark.streaming.kafka.KafkaCluster$$anonfun$2$$anonfun$3$$anonfun$apply$6$$anonfun$apply$7.apply(KafkaCluster.scala:90)
at scala.Option.map(Option.scala:146)
at org.apache.spark.streaming.kafka.KafkaCluster$$anonfun$2$$anonfun$3$$anonfun$apply$6.apply(KafkaCluster.scala:90)
at org.apache.spark.streaming.kafka.KafkaCluster$$anonfun$2$$anonfun$3$$anonfun$apply$6.apply(KafkaCluster.scala:87)
at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:241)
at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:241)
at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
at scala.collection.mutable.WrappedArray.foreach(WrappedArray.scala:35)
at scala.collection.TraversableLike$class.flatMap(TraversableLike.scala:241)
at scala.collection.AbstractTraversable.flatMap(Traversable.scala:104)
at org.apache.spark.streaming.kafka.KafkaCluster$$anonfun$2$$anonfun$3.apply(KafkaCluster.scala:87)
at org.apache.spark.streaming.kafka.KafkaCluster$$anonfun$2$$anonfun$3.apply(KafkaCluster.scala:86)
at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:241)
at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:241)
at scala.collection.immutable.Set$Set1.foreach(Set.scala:94)
at scala.collection.TraversableLike$class.flatMap(TraversableLike.scala:241)
at scala.collection.AbstractTraversable.flatMap(Traversable.scala:104)
at org.apache.spark.streaming.kafka.KafkaCluster$$anonfun$2.apply(KafkaCluster.scala:86)
at org.apache.spark.streaming.kafka.KafkaCluster$$anonfun$2.apply(KafkaCluster.scala:85)
at scala.util.Either$RightProjection.flatMap(Either.scala:522)
at org.apache.spark.streaming.kafka.KafkaCluster.findLeaders(KafkaCluster.scala:85)
at org.apache.spark.streaming.kafka.KafkaCluster.getLeaderOffsets(KafkaCluster.scala:179)
at org.apache.spark.streaming.kafka.KafkaCluster.getLeaderOffsets(KafkaCluster.scala:161)
at org.apache.spark.streaming.kafka.KafkaCluster.getEarliestLeaderOffsets(KafkaCluster.scala:155)
at org.apache.spark.streaming.kafka.KafkaUtils$$anonfun$5.apply(KafkaUtils.scala:213)
at org.apache.spark.streaming.kafka.KafkaUtils$$anonfun$5.apply(KafkaUtils.scala:211)
at scala.util.Either$RightProjection.flatMap(Either.scala:522)
at org.apache.spark.streaming.kafka.KafkaUtils$.getFromOffsets(KafkaUtils.scala:211)
at org.apache.spark.streaming.kafka.KafkaUtils$.createDirectStream(KafkaUtils.scala:484)
at SparkStreaming$.delayedEndpoint$SparkStreaming$1(SparkStreaming.scala:32)
at SparkStreaming$delayedInit$body.apply(SparkStreaming.scala:24)
at scala.Function0$class.apply$mcV$sp(Function0.scala:34)
at scala.runtime.AbstractFunction0.apply$mcV$sp(AbstractFunction0.scala:12)
at scala.App$$anonfun$main$1.apply(App.scala:76)
at scala.App$$anonfun$main$1.apply(App.scala:76)
at scala.collection.immutable.List.foreach(List.scala:381)
at scala.collection.generic.TraversableForwarder$class.foreach(TraversableForwarder.scala:35)
at scala.App$class.main(App.scala:76)
at SparkStreaming$.main(SparkStreaming.scala:24)
at SparkStreaming.main(SparkStreaming.scala)
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:497)
at com.intellij.rt.execution.application.AppMain.main(AppMain.java:144)

这些是我的依赖项

"org.apache.spark" %% "spark-streaming-kafka" % "1.6.2",
"org.apache.spark" %% "spark-core" % "1.6.2",
"org.apache.spark" % "spark-streaming_2.11" % "1.6.2",
"org.apache.kafka" %% "kafka" % "0.9.0.1"

我看不出问题出在哪里?有人可以帮我吗?

I can't see where the problem is? can anyone help me please?

推荐答案

根据Spark Streaming文档

According to Spark Streaming documentation here, Spark Streaming 1.6.2 is compatible with Kakfa 0.8.2.1.

Kafka::Spark Streaming 1.6.2与Kafka 0.8.2.1兼容

Kafka: Spark Streaming 1.6.2 is compatible with Kafka 0.8.2.1

因此，要解决您的问题，请使用版本0.8.2.1而不是0.9.0.1的kafka库.

So to solve your issue use kafka libraries of version 0.8.2.1 instead of 0.9.0.1.

希望这会有所帮助！

这篇关于kafka.cluster.BrokerEndPoint无法转换为kafka.cluster.Broker的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！