Spark Structured Streaming对接Kafka样例程序（Scala）

功能介绍

使用Structured Streaming，从kafka中读取广告请求数据、广告展示数据、广告点击数据，实时获取广告有效展示统计数据和广告有效点击统计数据，将统计结果写入kafka中。

代码样例

下面代码片段仅为演示，具体代码参见：com.huawei.bigdata.spark.examples.KafkaADCount。

/**
  * 运行Structured Streaming任务，统计广告的有效展示和有效点击数据，结果写入kafka中
  */

object KafkaADCount {
   def main(args: Array[String]): Unit = {
     if (args.length < 12) {
       System.err.println("Usage: KafkaWordCount <bootstrap-servers> " +
         "<maxEventDelay> <reqTopic> <showTopic> <maxShowDelay> " +
         "<clickTopic> <maxClickDelay> <triggerInterver> " +
         "<checkpointLocation> <protocol> <service> <domain>")
       System.exit(1)
     }

     val Array(bootstrapServers, maxEventDelay, reqTopic, showTopic,
     maxShowDelay, clickTopic, maxClickDelay, triggerInterver, checkpointLocation,
     protocol, service, domain) = args

     val maxEventDelayMills = JavaUtils.timeStringAs(maxEventDelay, TimeUnit.MILLISECONDS)
     val maxShowDelayMills = JavaUtils.timeStringAs(maxShowDelay, TimeUnit.MILLISECONDS)
     val maxClickDelayMills = JavaUtils.timeStringAs(maxClickDelay, TimeUnit.MILLISECONDS)
     val triggerMills = JavaUtils.timeStringAs(triggerInterver, TimeUnit.MILLISECONDS)

     val spark = SparkSession
       .builder
       .appName("KafkaADCount")
       .getOrCreate()

     spark.conf.set("spark.sql.streaming.checkpointLocation", checkpointLocation)

     import spark.implicits._

     // Create DataSet representing the stream of input lines from kafka
     val reqDf = spark
       .readStream
       .format("kafka")
       .option("kafka.bootstrap.servers", bootstrapServers)
       .option("kafka.security.protocol", protocol)
       .option("kafka.sasl.kerberos.service.name", service)
       .option("kafka.kerberos.domain.name", domain)
       .option("subscribe", reqTopic)
       .load()
       .selectExpr("CAST(value AS STRING)")
       .as[String]
       .map{
         _.split('^') match {
           case Array(reqAdID, reqTime) => ReqEvent(reqAdID,
             Timestamp.valueOf(reqTime))
         }
       }
       .as[ReqEvent]
       .withWatermark("reqTime", maxEventDelayMills +
         maxShowDelayMills + " millisecond")

     val showDf = spark
       .readStream
       .format("kafka")
       .option("kafka.bootstrap.servers", bootstrapServers)
       .option("kafka.security.protocol", protocol)
       .option("kafka.sasl.kerberos.service.name", service)
       .option("kafka.kerberos.domain.name", domain)
       .option("subscribe", showTopic)
       .load()
       .selectExpr("CAST(value AS STRING)")
       .as[String]
       .map{
         _.split('^') match {
           case Array(showAdID, showID, showTime) => ShowEvent(showAdID,
             showID, Timestamp.valueOf(showTime))
         }
       }
       .as[ShowEvent]
       .withWatermark("showTime", maxEventDelayMills +
         maxShowDelayMills + maxClickDelayMills + " millisecond")

     val clickDf = spark
       .readStream
       .format("kafka")
       .option("kafka.bootstrap.servers", bootstrapServers)
       .option("kafka.security.protocol", protocol)
       .option("kafka.sasl.kerberos.service.name", service)
       .option("kafka.kerberos.domain.name", domain)
       .option("subscribe", clickTopic)
       .load()
       .selectExpr("CAST(value AS STRING)")
       .as[String]
       .map{
         _.split('^') match {
           case Array(clickAdID, clickShowID, clickTime) => ClickEvent(clickAdID,
             clickShowID, Timestamp.valueOf(clickTime))
         }
       }
       .as[ClickEvent]
       .withWatermark("clickTime", maxEventDelayMills + " millisecond")

     val showStaticsQuery = reqDf.join(showDf,
       expr(s"""
       reqAdID = showAdID
       AND showTime >= reqTime + interval ${maxShowDelayMills} millisecond
       """))
       .selectExpr("concat_ws('^', showAdID, showID, showTime) as value")
       .writeStream
       .queryName("showEventStatics")
       .outputMode("append")
       .trigger(Trigger.ProcessingTime(triggerMills.millis))
       .format("kafka")
       .option("kafka.bootstrap.servers", bootstrapServers)
       .option("kafka.security.protocol", protocol)
       .option("kafka.sasl.kerberos.service.name", service)
       .option("kafka.kerberos.domain.name", domain)
       .option("topic", "showEventStatics")
       .start()

     val clickStaticsQuery = showDf.join(clickDf,
       expr(s"""
       showAdID = clickAdID AND
       showID = clickShowID AND
       clickTime >= showTime + interval ${maxClickDelayMills} millisecond
       """), joinType = "rightouter")
       .dropDuplicates("showAdID")
       .selectExpr("concat_ws('^', clickAdID, clickShowID, clickTime) as value")
       .writeStream
       .queryName("clickEventStatics")
       .outputMode("append")
       .trigger(Trigger.ProcessingTime(triggerMills.millis))
       .format("kafka")
       .option("kafka.bootstrap.servers", bootstrapServers)
       .option("kafka.security.protocol", protocol)
       .option("kafka.sasl.kerberos.service.name", service)
       .option("kafka.kerberos.domain.name", domain)
       .option("topic", "clickEventStatics")
       .start()

     new Thread(new Runnable {
       override def run(): Unit = {
         while(true) {
           println("-------------get showStatic progress---------")
           //println(showStaticsQuery.lastProgress)
           println(showStaticsQuery.status)
           println("-------------get clickStatic progress---------")
           //println(clickStaticsQuery.lastProgress)
           println(clickStaticsQuery.status)
           Thread.sleep(10000)
         }
       }
     }).start

     spark.streams.awaitAnyTermination()

   }
 }