在之前的文章中我们使用 Flink 也实现过 topn 的案例；这里，为了温习 Spark 如何访问 kafka 以及 DStream 的操作，我们实现一个需求：

需求：每天每地区热门广告 top3

一、数据源

[root@cdh101 kafka]# bin/kafka-console-consumer.sh --bootstrap-server cdh101:9092,cdh102:9092,cdh103:9092 --topic luchangyin --from-beginning

image.png

前置-引入依赖

<dependencies>

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.11</artifactId>
            <version>2.1.1</version>
        </dependency>

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming_2.11</artifactId>
            <version>2.1.1</version>
        </dependency>

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming-kafka-0-10_2.11</artifactId>
            <version>2.1.1</version>
        </dependency>

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_2.11</artifactId>
            <version>2.1.1</version>
        </dependency>

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-hive_2.11</artifactId>
            <version>2.1.1</version>
        </dependency>

    </dependencies>

二、代码实现

2.1 消费原始数据 ：

import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}

import java.text.SimpleDateFormat
import java.util.Date

// 需求：每天每地区热门广告 top3
object RealTime_App01 {

  def main(args: Array[String]): Unit = {

    //创建配置文件对象
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("HighKafka")
    //创建SparkStreaming执行的上下文
    val ssc = new StreamingContext(conf, Seconds(3))

    //kafka参数声明
    val brokers = "cdh101:9092,cdh102:9092,cdh103:9092"
    val topic = "luchangyin"
    val group = "cloudera_mirrormaker"
    val deserialization = "org.apache.kafka.common.serialization.StringDeserializer"
    val autooffsetreset = "latest"
    val kafkaParams = Map(
      ConsumerConfig.GROUP_ID_CONFIG -> group,
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> brokers,
      ConsumerConfig.AUTO_OFFSET_RESET_DOC -> autooffsetreset,
      ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> deserialization,
      ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> deserialization
    )

    //设置检查点目录
    ssc.checkpoint("D:\\MySoftware\\StudySoftware\\MyIdea\\luchangyin2021\\MyFirstBigScreen\\TestFSLJavaDemon\\src\\main\\ck1")

    //创建DS
    val kafkaDS: InputDStream[ConsumerRecord[String,String]] = KafkaUtils.createDirectStream[String,String](
        ssc,
        LocationStrategies.PreferConsistent,
        ConsumerStrategies.Subscribe[String,String](Set(topic), kafkaParams)
    )

    //从kafka的kv值中取value     1616683286749,华东,上海,102,1
    val dataDS = kafkaDS.map(_.value())
    dataDS.print()
    
    # 2.2 从kafka获取到的原始数据进行转换  ==>(天_地区_广告,1)

    ssc.start()
    ssc.awaitTermination()
  }

}

image.png

2.2 从kafka获取到的原始数据进行转换 ：

  val mapDS: DStream[(String, Int)] =dataDS.map{
      line => {
        val fields: Array[String] = line.split(",")
        //获取时间戳
        val timeStamp: Long = fields(0).toLong
        //根据时间戳创建日期对象
        val day = new Date(timeStamp)
        //创建SimpleDataFormat，对日期对象进行转换
        val sdf = new SimpleDateFormat("yyyy-MM-dd")
        //将日期对象转换为字符串
        val dayStr: String = sdf.format(day)
        //获取地区
        var area = fields(1)
        // 获取广告
        var adv = fields(4)
        //封装元组
        (dayStr +"_"+ area +"_"+ adv, 1)
      }
    }

    mapDS.print()  //(2021-03-25_华东_4,1)

    # 2.3 对每天每地区广告点击数进行聚合处理   (天_地区_广告,sum)

image.png

2.3 对每天每地区广告点击数进行聚合处理：

    //对每天每地区广告点击数进行聚合处理   (天_地区_广告,sum)
    //注意：这里要统计的是一天的数据，所以要将每一个采集周期的数据都统计，需要传递状态，所以要用udpateStateByKey
    val updateDS: DStream[(String, Int)] = mapDS.updateStateByKey(
      (seq: Seq[Int], buffer: Option[Int]) => {
        Option(seq.sum + buffer.getOrElse(0))
      }
    )

    updateDS.print()  //(2021-03-25_华东_3,138)
    
    # 2.4 将相同的天和地区放到一组

image.png

2.4 将相同的天和地区放到一组 ：

    //再次对结构进行转换
    val mapDS1: DStream[(String, (String, Int))] = updateDS.map{
      case (k, sum) => {
        val fields: Array[String] = k.split("_")
        (fields(0) +"_"+ fields(1), (fields(2), sum))
      }
    }

    //mapDS1.print() //  (2021-03-25_华北,(1,98))

    //将相同的天和地区放到一组
    val groupDS: DStream[(String, Iterable[(String, Int)])] = mapDS1.groupByKey()
    groupDS.print() //(2021-03-25_华中,ArrayBuffer((1,18), (2,21), (3,20), (4,22), (5,23)))

  # 2.5 对分组中的数据进行排序

image.png

2.5 对分组中的数据进行排序：

    // 对分组中的数据进行排序
    val resDS: DStream[(String, List[(String, Int)])] = groupDS.mapValues{
      datas => {
        datas.toList.sortBy(-_._2).take(3)
      }
    }

    // 打印结果
    resDS.print() // (2021-03-25_华北,List((5,107), (1,96), (3,92)))

image.png

   通过这个案例，我们晓得了 Spark 在项目中的应用，看代码感觉 scala 还是比 Java 简洁了好多，在实际Spark 的项目中 scala 还是占主导地位的，当然这也要看公司的规范以及个人的习惯了；好了，废话不多说了，多看几遍多敲几遍都会慢慢熟悉的，实践出真知，come on ， 少年 ~