Spark Streaming + Kafka Интеграция 0.8.2.1 - PullRequest
Новая
       9

Spark Streaming + Kafka Интеграция 0.8.2.1

0 голосов
/ 14 января 2019

У меня проблемы с интеграцией искры с кафкой. Я использовал spark-streaming-kafka-0-8. Я компилирую с SBT. Это мой код: 

import org.apache.spark.SparkConf
import org.apache.spark.streaming._
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.kafka._
object sparkKafka {

    def main(args: Array[String]) {

        val sparkConf = new SparkConf().setAppName("KafkaWordCount").setMaster("local[*]")

        val ssc = new StreamingContext(sparkConf, Seconds(2))

        val kafkaStream = KafkaUtils.createStream(ssc,
    "localhost:2181", "spark stream",  Map("customer" -> 2))

        kafkaStream.print()
        ssc.start()
        ssc.awaitTermination()
    }
}

Я получил эту ошибку: 

`[info] Running sparkKafka
[error] (run-main-0) java.lang.NoClassDefFoundError: scala/Product$class
[error] java.lang.NoClassDefFoundError: scala/Product$class
[error]         at org.apache.spark.SparkConf$DeprecatedConfig.<init>(SparkConf.scala:723)
[error]         at org.apache.spark.SparkConf$.<init>(SparkConf.scala:571)
[error]         at org.apache.spark.SparkConf$.<clinit>(SparkConf.scala)
[error]         at org.apache.spark.SparkConf.set(SparkConf.scala:92)
[error]         at org.apache.spark.SparkConf.set(SparkConf.scala:81)
[error]         at org.apache.spark.SparkConf.setAppName(SparkConf.scala:118)
[error]         at sparkKafka$.main(sparkKafka.scala:15)
[error]         at sparkKafka.main(sparkKafka.scala)
[error]         at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
[error]         at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
[error]         at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
[error]         at java.lang.reflect.Method.invoke(Method.java:498)
[error] Caused by: java.lang.ClassNotFoundException: scala.Product$class
[error]         at java.net.URLClassLoader.findClass(URLClassLoader.java:382)
[error]         at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
[error]         at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
[error]         at org.apache.spark.SparkConf$DeprecatedConfig.<init>(SparkConf.scala:723)
[error]         at org.apache.spark.SparkConf$.<init>(SparkConf.scala:571)
[error]         at org.apache.spark.SparkConf$.<clinit>(SparkConf.scala)
[error]         at org.apache.spark.SparkConf.set(SparkConf.scala:92)
[error]         at org.apache.spark.SparkConf.set(SparkConf.scala:81)
[error]         at org.apache.spark.SparkConf.setAppName(SparkConf.scala:118)
[error]         at sparkKafka$.main(sparkKafka.scala:15)
[error]         at sparkKafka.main(sparkKafka.scala)
[error]         at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
 [error]         at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
 [error]         at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
[error]         at java.lang.reflect.Method.invoke(Method.java:498)
[error] Nonzero exit code: 1
[error] (Compile / run) Nonzero exit code: 1
[error] Total time: 6 s, completed Jan 14, 2019 2:19:15 PM.`

Это мой файл build.sbt:

libraryDependencies += "org.apache.spark" % "spark-core_2.11" % "2.2.0" libraryDependencies += "org.apache.spark" % "spark-streaming_2.11" % "2.2.0" libraryDependencies += "org.apache.spark" % "spark-streaming-kafka-0-10_2.11" % "2.2.0" libraryDependencies += "org.apache.spark" % "spark-streaming-kafka-0-8_2.11" % "2.2.0"

Как я могу подключиться к Kafka? У меня проблема даже в spark-streaming-kafka-0-10 ....

Спасибо

1 Ответ

0 голосов
/ 14 января 2019

Это проблема версии со Scala или Spark. Убедитесь, что вы сначала используете Scala 2.11

Если вы используете Kafka 0.10 или выше (что, если вы недавно установили Kafka и используете его только локально, то, скорее всего, так и будет), то вам не следует использовать пакет kafka-0-8.

Не смешивать spark-streaming-kafka-0-8 с spark-streaming-kafka-0-10

Итак, если вы хотите использовать 0-10, , как уже было сказано ранее , пакет должен быть org.apache.spark.streaming.kafka010, а не org.apache.spark.streaming.kafka

Также обратите внимание, что 0-8 использует Zookeeper (например, localhost:2181), а 0-10 - нет.

...