卡夫卡消费者无法订阅卡夫卡主题(通过火花流运行)

Question

创建props对象后设置使用者的代码

val consumer = new KafkaConsumer[String, String](props)
consumer.subscribe(util.Arrays.asList(topic))

代码导入如下所示

package main.scala
import org.apache.spark.SparkContext
import org.apache.spark.SparkConf
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.Seconds
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.kafka.clients.consumer.KafkaConsumer
import java.util
import java.util.Properties
import org.apache.kafka.clients.consumer.{ConsumerConfig, KafkaConsumer}
import java.io.IOException

我通过sbt创建了一个程序集jar

libraryDependencies += "org.apache.spark" %% "spark-core" % "1.6.0" % "provided" 
libraryDependencies += "org.apache.spark" %% "spark-streaming" % "1.6.0" % "provided" 
libraryDependencies += "org.apache.spark" %% "spark-streaming-kafka" % "1.6.0" libraryDependencies += "org.apache.kafka" % "kafka_2.10" % "0.10.0-kafka-2.1.1"

我在这里错过了什么？

错误消息：

User类抛出异常: org.apache.kafka.clients.consumer.KafkaConsumer.subscribe(Ljava/util/Collection;)V :java.lang.NoSuchMethodError

Answer 1

我在Spark2.2.0和kafka 0.10.0中也有同样的问题，问题是因为在Spark2-mit(也是spak2-shell)中，kafka的默认版本不同。

我找到了决定这里

1. Before spark2-submit you have to export kafka version
$ export SPARK_KAFKA_VERSION=0.10
$ spark2-submit ...

Answer 2

subscribe采用java.util.Collections类型的输入，而不是java.util.Arrays.asList类型。

试一试

consumer.subscribe(java.util.Arrays.asList("topic"))

应该管用的..。