Shulou(Shulou.com)06/01 Report--

Spark2.3.1+Kafka0.9 uses Direct mode to consume abnormal information. I believe many inexperienced people don't know what to do about it. Therefore, this paper summarizes the causes and solutions of the problem. Through this article, I hope you can solve this problem.

Spark2.3.1+Kafka uses Direct mode consumption information Maven depends on org.apache.spark spark-streaming-kafka-0-8 customers 2.11 2.3.1 org.apache.spark spark-streaming_2.11 2.3.1

2.3.1 that is, spark version

Direct mode code import kafka.serializer.StringDecoderimport org.apache.spark.streaming.kafka.KafkaUtilsimport 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 does not support latest because the kafka version is too low.

Exception information Caused by: java.lang.NoSuchMethodException: scala.runtime.Nothing$. (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. (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 solution

You cannot use the default class of scala when validating the kafka property. You need to specify the class that kafka takes.

CreateDirectStream [String, String, StringDecoder, StringDecoder] where StringDecoder must be kafka.serializer.StringDecoder

After reading the above, have you mastered how Spark2.3.1+Kafka0.9 uses Direct mode to consume abnormal information? If you want to learn more skills or want to know more about it, you are welcome to follow the industry information channel, thank you for reading!

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.