Spark Streaming Kafka stream

Question

Mam pewne problemy podczas próby czytania z kafki z strumieniowaniem iskier.

Mój kod to:

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) 

ja wcześniej rozpoczął Zookeeper na porcie 2181 i serwerem Kafka 0.9.0.0 na porcie 9092. Ale pojawia się następujący błąd w sterowniku Spark:

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) 

Zookeeper dziennika:

[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) 

Każda podpowiedź?

Dziękuję bardzo

Answer 1

Problem związany był złą wersję zapłonie strumieniowego-Kafki.

Jak opisano w documentation

Kafki: Spark Streaming 1.5.2 jest kompatybilny z Kafki 0.8.2.1

Więc tym

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

w moim pom.xml (zamiast wersji 0.9.0.0) rozwiązało problem.

Nadzieja to pomaga

Answer 2

Kafka10 Streaming/Spark 2.1.0/DCOS/mezosferze

UGG spędziłem cały dzień na ten temat i musi przeczytać ten post kilkanaście razy. Próbowałem iskry 2.0.0, 2.0.1, Kafka 8, Kafka 10. Trzymaj się z daleka od Kafki 8 i iskry 2.0.x, a zależności są wszystkim. Zacznij od poniżej. To działa.

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" 

robocza Kafka/Spark kod Streaming:

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() 

proszę jak jeśli widzisz to tak mogę uzyskać kilka punktów reputacji. :)