我们这次 Spark-sql 操作中所有的数据均来自Hive，首先在Hive 中创建表,，并导入数据。
一共有3 张表： 1 张用户行为表，1 张城市表，1 张产品表

需求：各区域热门商品 Top3

这里的热门商品是从点击量的维度来看的，计算各个区域前三大热门商品，并备注上每
个商品在主要城市中的分布比例，超过两个城市用其他显示。
例如：
image.png

需求分析

➢查询出来所有的点击记录，并与city_info 表连接得到每个城市所在的地区 与Product_info 表连接得到产品名称
➢按照 地区和商品id 分组，统计出每个商品在每个地区的总点击次数
➢ 每个地区内按照点击次数降序排列
➢只取前三名
➢城市备注需要自定义 UDAF 函数

功能实现

➢ 连接三张表的数据，获取完整的数据（只有点击）
➢ 将数据根据地区，商品名称分组
➢ 统计商品点击次数总和,取 Top3
➢ 实现自定义聚合函数显示备注

代码

创建表

  1. package com.yang.bigdata.spark.sql
  3. import org.apache.spark.sql._
  6. /**
  7.  * @author ：ywb
  8.  * @date ：Created in 2022/1/23 6:10 下午
  9.  * @description：toDo
  10.  */
  11. object Spark06_SparkSQK_createTable {
  12.   def main(args: Array[String]): Unit = {
  15.     val spark: SparkSession = SparkSession
  16.       .builder()
  17.       .master("local[*]")
  18.       .appName("sqlTest")
  19.       .enableHiveSupport()
  20.       .getOrCreate()
  22.     //spark.sql("create database yang")
  24.     spark.sql("use yang")
  27.     //准备数据
  28.     spark.sql(
  29.       """
  30.         |CREATE TABLE `user_visit_action`(
  31.         |`date` string,
  32.         |`user_id` bigint,
  33.         |`session_id` string,
  34.         |`page_id` bigint,
  35.         |`action_time` string,
  36.         |`search_keyword` string,
  37.         |`click_category_id` bigint,
  38.         |`click_product_id` bigint,
  39.         |`order_category_ids` string,
  40.         |`order_product_ids` string,
  41.         |`pay_category_ids` string,
  42.         |`pay_product_ids` string,
  43.         |`city_id` bigint)
  44.         |row format delimited fields terminated  by '\t'
  45.         """.stripMargin)
  46.       spark.sql(
  47.         """
  48.           |load data local inpath 'datas/user_visit_action.txt' into table yang.user_visit_action
  49.           |""".stripMargin)
  51.     spark.sql(
  52.       """
  53.         |CREATE TABLE `product_info`(
  54.         |`product_id` bigint,
  55.         |`product_name` string,
  56.         |`extend_info` string)
  57.         |row format delimited fields terminated by '\t'
  58.         |""".stripMargin)
  60.     spark.sql(
  61.       """
  62.         |load data local inpath 'datas/product_info.txt' into table yang.product_info
  63.         |""".stripMargin)
  65.     spark.sql(
  66.       """
  67.         |CREATE TABLE `city_info`(
  68.         |`city_id` bigint,
  69.         |`city_name` string,
  70.         |`area` string)
  71.         |row format delimited fields terminated by '\t'
  72.         |""".stripMargin)
  74.     spark.sql(
  75.       """
  76.         |load data local inpath 'datas/city_info.txt' into table yang.city_info
  77.         |""".stripMargin)
  79.     spark.sql("""select * from city_info""").show()
  81.     spark.close()
  82.   }
  85. }
  1. object Spark06_SparkSQK_TODO1 {
  2.   def main(args: Array[String]): Unit = {
  4.     val spark: SparkSession = SparkSession
  5.       .builder()
  6.       .master("local[*]")
  7.       .appName("sqlTest")
  8.       .enableHiveSupport()
  9.       .getOrCreate()
  12.     spark.sql("use yang")
  15.     spark.sql(
  16.       """
  17.         |select
  18.         |    *
  19.         |from (
  20.         |    select
  21.         |        *,
  22.         |        rank() over(partition by area order by clickCount desc) as rank
  23.         |    from (
  24.         |        select
  25.         |            area,
  26.         |            product_name,
  27.         |            count(*) as clickCount
  28.         |        from (
  29.         |            select
  30.         |                u.*,
  31.         |                p.product_name,
  32.         |                c.area,
  33.         |                c.city_name
  34.         |            from user_visit_action u
  35.         |            join product_info p on u.click_product_id = p.product_id
  36.         |            join city_info c on u.city_id = c.city_id
  37.         |            where u.click_product_id > -1
  38.         |        ) t1 group by area,product_name
  39.         |    ) t2
  40.         |) t3 where rank <= 3
  41.         |""".stripMargin).show()
  43.     spark.close()
  44.   }
  47. }
  1. package com.yang.bigdata.spark.sql
  3. import org.apache.commons.lang.mutable.Mutable
  4. import org.apache.spark.sql._
  5. import org.apache.spark.sql.expressions.Aggregator
  7. import scala.collection.mutable
  8. import scala.collection.mutable.ListBuffer
  11. /**
  12.  * @author ：ywb
  13.  * @date ：Created in 2022/1/23 6:10 下午
  14.  * @description：toDo
  15.  */
  16. object Spark06_SparkSQK_TODO2 {
  17.   def main(args: Array[String]): Unit = {
  18.     val spark: SparkSession = SparkSession
  19.       .builder()
  20.       .master("local[*]")
  21.       .appName("sqlTest2")
  22.       .enableHiveSupport()
  23.       .getOrCreate()
  25.     spark.sql("use yang")
  27.     spark.sql(
  28.       """
  29.         |select
  30.         |     a.*,
  31.         |     p.product_name,
  32.         |     c.area,
  33.         |     c.city_name
  34.         |from user_visit_action a
  35.         |join product_info p on a.click_product_id = p.product_id
  36.         |join city_info c on a.city_id = c.city_id
  37.         |where a.click_product_id > -1
  38.         |""".stripMargin).createOrReplaceTempView("t1")
  40.     spark.udf.register("dsc", functions.udaf(new DescAgg))
  42.     spark.sql(
  43.       """
  44.         |select
  45.         |     area,
  46.         |     product_name,
  47.         |     count(*) as click_count,
  48.         |     dsc(city_name)
  49.         |from t1
  50.         |group by area,product_name
  51.         |""".stripMargin).createOrReplaceTempView("t2")
  53.     spark.sql(
  54.       """
  55.         |select
  56.         |     *,
  57.         |     rank() over(partition by area order by click_count desc) as rank
  58.         |from t2
  59.         |""".stripMargin).createOrReplaceTempView("t3")
  61.     spark.sql(
  62.       """
  63.         |select
  64.         |     *
  65.         |from t3
  66.         |where rank <= 3
  67.         |""".stripMargin).show()
  69.     spark.close()
  70.   }
  71.   // countSum ==》 [(cityName,count),(cityName,count) ...... ]
  72.   case class Buff( var sum : Long , var map : mutable.Map[String,Long] )
  74.   class DescAgg extends Aggregator[String,Buff,String]{
  76.         //初始化
  77.         override def zero: Buff = {
  78.             Buff( 0L , mutable.Map[String,Long]())
  79.         }
  81.         //聚合
  82.         override def reduce(buff: Buff, cityName: String): Buff = {
  83.             buff.sum += 1L
  84.             val count: Long = buff.map.getOrElse(cityName, 0L) + 1
  85.             buff.map.update(cityName,count)
  86.             buff
  87.         }
  89.         //合并
  90.         override def merge(b1: Buff, b2: Buff): Buff = {
  91.           val buffMap1: mutable.Map[String, Long] = b1.map
  92.           val buffMap2: mutable.Map[String, Long] = b2.map
  93.           buffMap2.foreach(
  94.              {
  95.                case (k,v) => {
  96.                  val count: Long = buffMap1.getOrElse(k, 0L) + v
  97.                  buffMap1.update(k,count)
  98.                }
  99.              }
  100.           )
  101.           b1.sum += b2.sum
  102.           b1.map = buffMap1
  103.           b1
  104.         }
  106.         // 最终值
  107.         override def finish(temp: Buff): String = {
  108.           val resList: ListBuffer[String] = ListBuffer[String]()
  110.           val sum: Long = temp.sum
  111.           val map: mutable.Map[String, Long] = temp.map
  113.           val list: List[(String, Long)] = map.toList
  114.           val hasMore: Boolean = list.size > 2
  115.           val res: List[(String, Long)] = list.sortBy(_._2).take(2)
  117.           var percentage: Long = 0L
  118.           res.foreach(
  119.             {
  120.               case (k,v) => {
  121.                   var p: Long = v * 100 / sum
  122.                   percentage += p
  123.                   resList.append(s"$k $p%")
  124.               }
  125.             }
  126.           )
  128.           if (hasMore){
  129.             resList.append(s"其他 ${100-percentage}%")
  130.           }
  131.           resList.mkString(",")
  132.         }
  134.         override def bufferEncoder: Encoder[Buff] = Encoders.product
  136.         override def outputEncoder: Encoder[String] = Encoders.STRING
  137.   }
  139. }