Sparkstreaming + Кафка в hdfs

Question

Когда я пытаюсь использовать сообщение из темы kafka, используя spark streaming, получаю ошибку ниже

scala> val kafkaStream = KafkaUtils.createStream(ssc, "<ipaddress>:2181","spark-streaming-consumer-group", Map("test1" -> 5))

Ошибка:

`missing or invalid dependency detected while loading class file 'KafkaUtils.class'.
Could not access term kafka in package <root>,
because it (or its dependencies) are missing. Check your build definition for
missing or conflicting dependencies. (Re-run with `-Ylog-classpath` to see the problematic classpath.)
A full rebuild may help if 'KafkaUtils.class' was compiled against an incompatible version of <root>.`

Версия Scala: 2.11.8 версия spark: 2.1.0.2.6.0.3-8

Я использовал все виды библиотек для spark-streaming-kafka, но ничего не получалось:

Я выполняю код из оболочки spark:

./spark-shell --jars /data/home/local/504/spark-streaming-kafka_2.10-1.5.1.jar, /data/home/local/504/spark-streaming_2.10-1.5.1.jar

Код

import org.apache.spark.SparkConf
val conf = new SparkConf().setMaster("local[*]").setAppName("KafkaReceiver")
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.Seconds
val ssc = new StreamingContext(conf, Seconds(10))
import org.apache.spark.streaming.kafka.KafkaUtils
val kafkaStream = KafkaUtils.createStream(ssc, "<ipaddress>:2181","spark-streaming-consumer-group", Map("test1" -> 5))

Любые предложения по этому вопросу.

Answer 1

Поскольку вы используете Scala 2.11 и spark 2.1.0, вы должны использовать эти банки

• spark-streaming-kafka-0-10_2.11-2.1.0.jar
• spark-streaming_2.11-2.1.0.jar

Если вы используете Kafka 0.10+, измените его соответствующим образом.

И простая программа будет выглядеть как

import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.kafka.common.serialization.StringDeserializer

val streamingContext = new StreamingContext(sc, Seconds(5))

//Parameters for kafka
val kafkaParams = Map[String, Object](
  "bootstrap.servers" -> "servers,
  "key.deserializer" -> classOf[StringDeserializer],
  "value.deserializer" -> classOf[StringDeserializer],
  "group.id" -> "test-consumer-group",
  "auto.offset.reset" -> "earliest",
  "enable.auto.commit" -> (false: java.lang.Boolean)
)
val topics = "topics,seperated,by,comma".split(",")

// crate dstreams
val stream = KafkaUtils.createDirectStream[String, String](
  streamingContext,
  PreferConsistent,
  Subscribe[String, String](topics, kafkaParams)
)

//stream.print()
stream.map(_.value().toString).print()

Надеюсь, это hepls!