七、案例实操1：各区域热门商品 Top3

1，数据说明

编号	字段名称	字段类型	字段含义
1	date	String	用户点击行为的日期
2	user_id	Long	用户的 ID
3	session_id	String	Session 的 ID
4	page_id	Long	某个页面的 ID
5	action_time	String	动作的时间点
6	search_keyword	String	用户搜索的关键词
7	click_category_id	Long	某一个商品品类的 ID
8	click_product_id	Long	某一个商品的 ID
9	order_category_ids	String	一次订单中所有品类的 ID 集合
10	order_product_ids	String	一次订单中所有商品的 ID 集合
11	pay_category_ids	String	一次支付中所有品类的 ID 集合
12	pay_product_ids	String	一次支付中所有商品的 ID 集合
13	city_id	Long	城市 id

• 完整数据：数据.zip

3，需求说明

地区	商品名称	点击次数	城市备注
华北	商品 A	100000	北京 21.2%，天津 13.2%，其他 65.6%
华北	商品 P	80200	北京 63.0%，太原 10%，其他 27.0%
华北	商品 M	40000	北京 63.0%，太原 10%，其他 27.0%
东北	商品 J	92000	大连 28%，辽宁 17.0%，其他 55.0%

（2）要实现该需求，大概思路如下：

4，功能实现

// 自定义聚合函数的缓冲区数据类型
case class Buffer( var total : Long, var cityMap:mutable.Map[String, Long] )

// 自定义聚合函数：实现城市备注功能
// 1. 继承Aggregator, 定义泛型
//    IN ： 城市名称
//    BUF : Buffer =>【总点击数量，Map[（city, cnt）, (city, cnt)]】
//    OUT : 备注信息
// 2. 重写6个方法
class CityRemarkUDAF extends Aggregator[String, Buffer, String]{
  // 缓冲区初始化
  override def zero: Buffer = {
    Buffer(0, mutable.Map[String, Long]())
  }

  // 更新缓冲区数据
  override def reduce(buff: Buffer, city: String): Buffer = {
    buff.total += 1
    val newCount = buff.cityMap.getOrElse(city, 0L) + 1
    buff.cityMap.update(city, newCount)
    buff
  }

  // 合并缓冲区数据
  override def merge(buff1: Buffer, buff2: Buffer): Buffer = {
    buff1.total += buff2.total

    val map1 = buff1.cityMap
    val map2 = buff2.cityMap

    // 两个Map的合并操作
    map2.foreach{
      case (city , cnt) => {
        val newCount = map1.getOrElse(city, 0L) + cnt
        map1.update(city, newCount)
      }
    }
    buff1.cityMap = map1
    buff1
  }

  // 将统计的结果生成字符串信息
  override def finish(buff: Buffer): String = {
    val remarkList = ListBuffer[String]()

    val totalcnt = buff.total
    val cityMap = buff.cityMap

    // 降序排列
    val cityCntList = cityMap.toList.sortWith(
      (left, right) => {
        left._2 > right._2
      }
    ).take(2)

    val hasMore = cityMap.size > 2
    var rsum = 0L
    cityCntList.foreach{
      case ( city, cnt ) => {
        val r = cnt * 100 / totalcnt
        remarkList.append(s"${city} ${r}%")
        rsum += r
      }
    }
    if ( hasMore ) {
      remarkList.append(s"其他 ${100 - rsum}%")
    }

    remarkList.mkString(", ")
  }

  // Spark 用于序列化缓冲区的编码器
  override def bufferEncoder: Encoder[Buffer] = Encoders.product

  // Spark 用于序列化输出结果的编码器
  override def outputEncoder: Encoder[String] = Encoders.STRING
}

（2）接着我们程序中通过如下步骤进行统计计算：

object Hello {
  def main(args: Array[String]): Unit = {
    // 创建 Spark 运行配置对象
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Hello")
    //创建 SparkSession 对象
    val spark: SparkSession = SparkSession.builder().config(sparkConf)
      .enableHiveSupport() // 添加Hive支持
      .getOrCreate()

    // 查询基本数据
    spark.sql(
      """
        |  select
        |     a.*,
        |     p.product_name,
        |     c.area,
        |     c.city_name
        |  from user_visit_action a
        |  join product_info p on a.click_product_id = p.product_id
        |  join city_info c on a.city_id = c.city_id
        |  where a.click_product_id > -1
            """.stripMargin).createOrReplaceTempView("t1")

    // 根据区域，商品进行数据聚合
    spark.udf.register("cityRemark", functions.udaf(new CityRemarkUDAF()))
    spark.sql(
      """
        |  select
        |     area,
        |     product_name,
        |     count(*) as clickCnt,
        |     cityRemark(city_name) as city_remark
        |  from t1 group by area, product_name
            """.stripMargin).createOrReplaceTempView("t2")

    // 区域内对点击数量进行排行
    spark.sql(
      """
        |  select
        |      *,
        |      rank() over( partition by area order by clickCnt desc ) as rank
        |  from t2
            """.stripMargin).createOrReplaceTempView("t3")

    // 取前3名
    spark.sql(
      """
        | select
        |     *
        | from t3 where rank <= 3
            """.stripMargin).show(false)

    //关闭 Spark
    spark.stop()
  }
}

（3）运行结果如下：