Spark Streaming项目实战(2) | 最近 1 小时广告点击量实时统计

约定不等于承诺〃 2022-12-04 11:58 14阅读 0赞

请先Spark Streaming项目实战(1)  
[https://blog.csdn.net/qq\_46548855/article/details/108436569][https_blog.csdn.net_qq_46548855_article_details_108436569]  
源码在github：[https://github.com/lidonglin-bit/Spark-Streaming][https_github.com_lidonglin-bit_Spark-Streaming]  
统计各广告最近 1 小时内的点击量趋势：各广告最近 1 小时内各分钟的点击量  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center]

*  分析怎么实现

统计各广告最近1个小时内的点击量趋势：各广告最近1个小时内分钟的点击量，每6秒统计一次
    1.各广告      ->          按照广告分钟
    2.最近1个小时，每6秒统计一次      ->     窗口：窗口长度1个小时   窗口的滑动步长5s
    -------------
    
    1.先把窗口分好
    2.按照广告分组，进行聚合
    3.按照广告分组，把这个广告下所有的分钟记录在一起

*  代码实现

import bean.AdsInfo
    import org.apache.spark.streaming.{ Minutes, Seconds}
    import org.apache.spark.streaming.dstream.DStream
    import org.json4s.jackson.JsonMethods
    import util.RedisUtil
    
    object LastHourApp extends App { 
      override def doSomething(adsInfoStream: DStream[AdsInfo]): Unit = { 
        adsInfoStream
          //1.把窗口分好
          .window(Minutes(60),Seconds(3))
        //2.按照广告分组，进行聚合
          .map(info=>((info.adsId,info.hmString),1))
          .reduceByKey(_+_)
        //((4,10:14),100)
         // .print(1000)
        //3.按照广告分组，把这个广告下所有的分钟记录在一起
          .map{ 
            case ((ads,hm),count)=>(ads,(hm,count))
          }
          .groupByKey()
        //.print(10000)
        //4。写入到redis
          .foreachRDD(rdd=>{ 
            rdd.foreachPartition((it: Iterator[(String, Iterable[(String, Int)])]) => { 
              if (it.nonEmpty){    //只是判断是否有下一个元素，指针不会跳过这个元素
                //先建立到redis连接
                val client = RedisUtil.getClient
                //2.写元素到redis
                //2.1 一个一个的写(需求1用)
                //2.2 批次写入(需求二用)
                import org.json4s.JsonDSL._
                val key = "last:ads:hour:count"
                val map = it.toMap.map{ 
                  case (adsId,it) => (adsId,JsonMethods.compact(JsonMethods.render(it)))
                }
                //sacla集合转换成java集合
                import scala.collection.JavaConversions._
                println(map)
                client.hmset(key,map)
    
                //3.关闭redis (用的是连接池，实际上是把连接归还给连接池)
                client.close()
              }
            })
          })
    
      }
    }

*  结果  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 1]  
    redis中  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 2]

[https_blog.csdn.net_qq_46548855_article_details_108436569]: https://blog.csdn.net/qq_46548855/article/details/108436569
[https_github.com_lidonglin-bit_Spark-Streaming]: https://github.com/lidonglin-bit/Spark-Streaming
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/a6795083049f433ab43665aa51a04861.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221123/17e70efeed0c4c0fb2d9cf0901ad0e47.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221123/29524f51691c466f8e60e434358b26f3.png