Spark2.3.1+Kafka0.9使用Direct模式消费信息异常怎么办，相信很多没有经验的人对此束手无策，为此本文总结了问题出现的原因和解决方法，通过这篇文章希望你能解决这个问题。
计算
安全
数据库
网络和加速
企业服务
Spark2.3.1+Kafka使用Direct模式消费信息Maven依赖<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming-kafka-0-8_2.11</artifactId>
<version>2.3.1</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming_2.11</artifactId>
<version>2.3.1</version>
</dependency>2.3.1即spark版本
Direct模式代码import kafka.serializer.StringDecoder
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}
object Test {
val zkQuorum = "mirrors.mucang.cn:2181"
val groupId = "nginx-cg"
val topic = Map("nginx-log" -> 1)
val KAFKA_INTERVAL = 10
case class NginxInof(domain: String, ip: String)
def main(args: Array[String]): Unit = {
val sparkConf = new SparkConf().setAppName("NginxLogAnalyze").setMaster("local[*]")
val sparkContext = new SparkContext(sparkConf)
val streamContext = new StreamingContext(sparkContext, Seconds(KAFKA_INTERVAL))
val kafkaParam = Map[String, String](
"bootstrap.servers" -> "xx.xx.cn:9092",
"group.id" -> "nginx-cg",
"auto.offset.reset" -> "largest"
)
val topic = Set("nginx-log")
val kafkaStream = KafkaUtils.createDirectStream(streamContext, kafkaParam, topic)
val counter = kafkaStream
.map(_.toString().split(" "))
.map(item => (item(0).split(",")(1) + "-" + item(2), 1))
.reduceByKey((x, y) => (x + y))
counter.foreachRDD(rdd => {
rdd.foreach(println)
})
streamContext.start()
streamContext.awaitTermination()
}
}largest 因为kafka版本过低不支持latest
Caused by: java.lang.NoSuchMethodException: scala.runtime.Nothing$.<init>(kafka.utils.VerifiableProperties)
at java.lang.Class.getConstructor0(Class.java:3082)
at java.lang.Class.getConstructor(Class.java:1825)
at org.apache.spark.streaming.kafka.KafkaRDD$KafkaRDDIterator.<init>(KafkaRDD.scala:153)
at org.apache.spark.streaming.kafka.KafkaRDD.compute(KafkaRDD.scala:136)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:96)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:53)
at org.apache.spark.scheduler.Task.run(Task.scala:109)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:345)
... 3 more在验证kafka属性时不能使用scala默认的类，需要指定kafka带的类createDirectStream[String, String, StringDecoder, StringDecoder]其中StringDecoder必须是kafka.serializer.StringDecoder
看完上述内容，你们掌握Spark2.3.1+Kafka0.9使用Direct模式消费信息异常怎么办的方法了吗？如果还想学到更多技能或想了解更多相关内容，欢迎关注亿速云行业资讯频道，感谢各位的阅读！
亿速云「云服务器」，即开即用、新一代英特尔至强铂金CPU、三副本存储NVMe SSD云盘，价格低至29元/月。点击查看>>
免责声明：本站发布的内容（图片、视频和文字）以原创、转载和分享为主，文章观点不代表本网站立场，如果涉及侵权请联系站长邮箱：is@yisu.com进行举报，并提供相关证据，一经查实，将立刻删除涉嫌侵权内容。
原文链接：https://my.oschina.net/u/4725115/blog/4930415
