Spark - SparkSQL使用详解7（案例实操1：各区域热门商品Top3）

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

1，数据说明

• 完整数据：数据.zip

（2）文件 product_info.txt 中则存放的是商品信息，下面是该文件的部分内容：

1	商品_1	自营
2	商品_2	自营
3	商品_3	第三方
4	商品_4	自营
5	商品_5	自营

（3）文件 city_info.txt 中则存放的是城市信息，下面是该文件的部分内容：

1	北京	华北
2	上海	华东
3	深圳	华南
4	广州	华南
5	武汉	华中
6	南京	华东
7	天津	华北

2，数据准备

（1）在实现具体需求之前，我们首先需要创建相应的 Hive 表，并将 txt 文件中的数据导入到表中，具体代码如下：

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()

    // 导入隐式转换
    import spark.implicits._

    // 创建3张表并导入初始化数据
    spark.sql(
      """
        |CREATE TABLE `user_visit_action`(
        |  `date` string,
        |  `user_id` bigint,
        |  `session_id` string,
        |  `page_id` bigint,
        |  `action_time` string,
        |  `search_keyword` string,
        |  `click_category_id` bigint,
        |  `click_product_id` bigint,
        |  `order_category_ids` string,
        |  `order_product_ids` string,
        |  `pay_category_ids` string,
        |  `pay_product_ids` string,
        |  `city_id` bigint)
        |row format delimited fields terminated by '\t'
      """.stripMargin)

    spark.sql(
      """
        |load data local inpath 'datas/user_visit_action.txt' into table user_visit_action
      """.stripMargin)

    spark.sql(
      """
        |CREATE TABLE `product_info`(
        |  `product_id` bigint,
        |  `product_name` string,
        |  `extend_info` string)
        |row format delimited fields terminated by '\t'
      """.stripMargin)

    spark.sql(
      """
        |load data local inpath 'datas/product_info.txt' into table product_info
      """.stripMargin)

    spark.sql(
      """
        |CREATE TABLE `city_info`(
        |  `city_id` bigint,
        |  `city_name` string,
        |  `area` string)
        |row format delimited fields terminated by '\t'
      """.stripMargin)

    spark.sql(
      """
        |load data local inpath 'datas/city_info.txt' into table city_info
      """.stripMargin)

    // 查看目前所有的Hive表
    spark.sql("show tables").show()

    // 查看其中 city_info 表的模式信息，包括列名、数据类型和其他元数据信息
    spark.sql("DESCRIBE user_visit_action").show()

    // 查看 city_info 表的数据
    spark.sql("""select * from city_info""").show

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

（2）运行结果如下说明数据已经准备完毕：

（3）由于我使用的是内置的 Hive，可以看到项目根目录下会出现一个 spark-warehouse 文件夹，该文件夹下的的每个子文件夹是一张表，子文件夹名为表名，里面则是表的数据文件。

3，需求说明

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
}

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()
  }
}