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

我使用 spark 1.6.2 和 kafka 0.9.0.1 和 scala 2.11.8

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 文档 此处，Spark Streaming 1.6.2 与 Kakfa 0.8.2.1 兼容.

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屋！