スパークストリーミングカフカストリーム

Question

スパークストリーミングでkafkaからの読み込み中に問題が発生しました。

私のコードは次のとおりです。

val sparkConf = new SparkConf().setMaster("local[2]").setAppName("KafkaIngestor") 
val ssc = new StreamingContext(sparkConf, Seconds(2)) 

val kafkaParams = Map[String, String](
    "zookeeper.connect" -> "localhost:2181", 
    "group.id" -> "consumergroup", 
    "metadata.broker.list" -> "localhost:9092", 
    "zookeeper.connection.timeout.ms" -> "10000" 
    //"kafka.auto.offset.reset" -> "smallest" 
) 

val topics = Set("test") 
val stream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics) 

私は以前ポート9092. でポート2181での飼育係とカフカサーバー0.9.0.0を開始した。しかし、私はスパークドライバで次のエラーを取得する：

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:145) 
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) 

飼育係ログ：

[2015-12-08 00:32:08,226] INFO Got user-level KeeperException when processing sessionid:0x1517ec89dfd0000 type:create cxid:0x34 zxid:0x1d3 txntype:-1 reqpath:n/a Error Path:/brokers/ids Error:KeeperErrorCode = NodeExists for /brokers/ids (org.apache.zookeeper.server.PrepRequestProcessor) 

任意のヒント？

は問題が間違った火花ストリーミング・カフカのバージョンを関連していた。非常に

Answer 1

をありがとうdocumentation

カフカで説明したように

：スパークは1.5.2をストリーミングは私のpom.xmlで

<dependency> 
    <groupId>org.apache.kafka</groupId> 
    <artifactId>kafka_2.10</artifactId> 
    <version>0.8.2.2</version> 
</dependency> 

含め、0.8.2.1

だから、カフカと互換性があります（バージョン0.9.0.0ではなく）この問題が解決されました。

希望これは、私はこの上のすべての一日を過ごしたと、このポストに12回を読んでいなければなりません

Answer 2

Kafka10ストリーミング/スパーク2.1.0/DCOS /中間圏

Uggのに役立ちます。私はspark 2.0.0、2.0.1、Kafka 8、Kafka 10.を試しました。Kafka 8から離れて2.0.xを起動し、依存関係はすべてです。まずは以下から始めてください。できます。

SBT：

"org.apache.hadoop" % "hadoop-aws" % "2.7.3" excludeAll ExclusionRule(organization = "org.apache.hadoop", name = "hadoop-common"), 
"org.apache.spark" %% "spark-core" % "2.1.0", 
"org.apache.spark" %% "spark-sql" % "2.1.0" , 
"org.apache.spark" % "spark-streaming-kafka-0-10_2.11" % "2.1.0", 
"org.apache.spark" % "spark-streaming_2.11" % "2.1.0" 

ワーキングカフカ/スパークストリーミングコード：

val spark = SparkSession 
    .builder() 
    .appName("ingest") 
    .master("local[4]") 
    .getOrCreate() 

import spark.implicits._ 
val ssc = new StreamingContext(spark.sparkContext, Seconds(2)) 

val topics = Set("water2").toSet 

val kafkaParams = Map[String, String](
    "metadata.broker.list"  -> "broker:port,broker:port", 
    "bootstrap.servers"   -> "broker:port,broker:port", 
    "group.id"     -> "somegroup", 
    "auto.commit.interval.ms"  -> "1000", 
    "key.deserializer"   -> "org.apache.kafka.common.serialization.StringDeserializer", 
    "value.deserializer"   -> "org.apache.kafka.common.serialization.StringDeserializer", 
    "auto.offset.reset"   -> "earliest", 
    "enable.auto.commit"   -> "true" 
) 

val messages = KafkaUtils.createDirectStream[String, String](ssc, PreferConsistent, Subscribe[String, String](topics, kafkaParams)) 

messages.foreachRDD(rdd => { 
    if (rdd.count() >= 1) { 
    rdd.map(record => (record.key, record.value)) 
     .toDS() 
     .withColumnRenamed("_2", "value") 
     .drop("_1") 
     .show(5, false) 
    println(rdd.getClass) 
    } 
}) 
ssc.start() 
ssc.awaitTermination() 

あなたがこれを見た場合のようなので、私は得ることができますしてくださいいくつかの評判ポイント。 :)