spark streaming kafka example

Posted 2020-06-17

// scalastyle:off println
package org.apache.spark.examples.streaming

import kafka.serializer.StringDecoder
import org.apache.spark.SparkConf
import org.apache.spark.streaming._
import org.apache.spark.streaming.kafka._
import org.apache.spark.streaming.scheduler.StreamingListener
import scala.util.parsing.json.JSON

/**
 * Consumes messages from one or more topics to analysis log
 * calaculate the threadhold under certain time window
 */
object LogAnalysisB {
  def main(args: Array[String]) {
    if (args.length < 2) {
      System.err.println(s"""
        |Usage: DirectKafkaWordCount <brokers> <topics>
        |  <brokers> is a list of one or more Kafka brokers
        |  <topics> is a list of one or more kafka topics to consume from
        |
        """.stripMargin)
      System.exit(1)
    }
    val WINDOW_LENGTH = new Duration(30 * 1000)
    val SLIDE_INTERVAL = new Duration(10 * 1000)
    StreamingExamples.setStreamingLogLevels()
    val Array(brokers, topics) = args
    val sparkConf = new SparkConf().setAppName("ELK Log Analysis windows Threhold")
    val ssc = new StreamingContext(sparkConf,SLIDE_INTERVAL)
    ssc.addStreamingListener(new RuleFileListenerB())
    // Create direct kafka stream with brokers and topics
    val topicsSet = topics.split(",").toSet
    val kafkaParams = Map[String, String]("metadata.broker.list" -> brokers)
    val messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
      ssc, kafkaParams, topicsSet)

    // Get the lines, split them into words, count the words and print
    val lines = messages.map(_._2).map(HostAppLog.parseLogLine)
    val windowDStream = lines.window(WINDOW_LENGTH,SLIDE_INTERVAL)
    windowDStream.foreachRDD( logs=>
    {
      val topChar = logs
        .map(log => (log.msg, 1))
        .reduceByKey(_ + _)
        .top(3)(OrderingUtils.SecondValueOrdering)
      println("$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$")
      println( s"""Top Endpoints: ${topChar.mkString("[", ",", "]")}""")

      val topTest = logs
      .map(log =>(log.host+log.app,if (log.msg.contains("A")) 1 else 0))
      .reduceByKey(_+_)
      .filter(_._2 > 5)
      .take(10)
      println( s"""A > 5 times: ${topTest.mkString("[", ",", "]")}""")
    }
    )

    // Start the computation
    ssc.start()
    ssc.awaitTermination()
  }
   def wc(ssc:StreamingContext,map:Map[Any,Any]): Unit =
   {
     if( map.get("message").toString().contains("A"))
       println("find A in message:" + map.toString())

   }

}

class RuleFileListenerB extends StreamingListener {

  override def onBatchStarted(batchStarted : org.apache.spark.streaming.scheduler.StreamingListenerBatchStarted) {
    println("---------------------------------------------------------------------------------------------------------------------------------------------")
        println("check whether the file‘s modified date is change, if change then reload the configuration file")
    //val source = scala.io.Source.fromFile("D:/code/scala/test")
    //val lines = try source.mkString finally source.close()
    //println(lines)
    println("---------------------------------------------------------------------------------------------------------------------------------------------")
  }

}
// scalastyle:on println