Spark Core项目实战(2) | Top10热门品类中每个品类的 Top10 活跃 Session 统计

古城微笑少年丶 2022-12-01 11:50 207阅读 0赞

数据源

链接：https://pan.baidu.com/s/1TtBQpQUNBebqxrrx9czxqQ 
    提取码：fmw6

源码在github：[https://github.com/lidonglin-bit/Spark-Core][https_github.com_lidonglin-bit_Spark-Core]  
**`根据上个需求的基础上求的Top10热门品类中每个品类的 Top10 活跃 Session 统计`**(建议先做实战1，再做实战2)  
[https://blog.csdn.net/qq\_46548855/article/details/108198209][https_blog.csdn.net_qq_46548855_article_details_108198209]

### 目录 ###

*  一.Top10热门品类中每个品类的 Top10 活跃 Session 统计
 *   *  需求分析
     *  思路
 *  二.具体实现
 *   *  具体业务实现
     *  完整项目代码
     *  3种方案的对比(建议用方案4)

# 一.Top10热门品类中每个品类的 Top10 活跃 Session 统计 #

## 需求分析 ##

对于排名前 10 的品类，分别获取每个品类点击次数排名前 10 的 sessionId。(注意: 这里我们只关注点击次数, 不关心下单和支付次数)  
这个就是说，对于 top10 的品类，每一个都要获取对它点击次数排名前 10 的 sessionId。  
这个功能，可以让我们看到，对某个用户群体最感兴趣的品类，各个品类最感兴趣最典型的用户的 session 的行为。

## 思路 ##

1.  过滤出来 category Top10的日志
2.  需要用到需求1的结果, 然后只需要得到categoryId就可以了
3.  转换结果为 RDD\[(categoryId, sessionId), 1\] 然后统计数量 => RDD\[(categoryId, sessionId), count\]
4.  统计每个品类 top10. => RDD\[categoryId, (sessionId, count)\] => RDD\[categoryId, Iterable\[(sessionId, count)\]\]
5.  对每个 Iterable\[(sessionId, count)\]进行排序, 并取每个Iterable的前10
6.  把数据封装到 CategorySession 中  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center]

# 二.具体实现 #

## 具体业务实现 ##

import java.util
    
    import bean.{ CategoryCountInfo, SessionInfo, UserVisitAction}
    import org.apache.spark.{ Partitioner, SparkContext}
    import org.apache.spark.rdd.RDD
    
    import scala.collection.mutable
    
    object CtegorySessionTopApp { 
      def statCategorySessionTop10(sc:SparkContext,categoryTop10: Array[CategoryCountInfo],
                                   UserVisitActionRDD: RDD[UserVisitAction]) = { 
        /* 解决方案1 */            
        //1.过滤出来，只包含top10品类的那些点击记录
        //1.1先把top10品类id拿出来,转成long型的id 目的是为了和UserVisitActionRDD接轨
        val cids = categoryTop10.map(_.categoryId.toLong)
        val filteredUserVisitActionRDD = UserVisitActionRDD.filter(action => cids.contains(action.click_category_id))
    
        //2.每个品类top10session的计算
        //2.1 先map需要字段
        val cidSidAndOne =
        filteredUserVisitActionRDD.map(action => ((action.click_category_id, action.session_id), 1))
        //2.2最聚合操作 RDD[(cid,sid),1))]
        val cidSidAndCount = cidSidAndOne.reduceByKey(_ + _)
        //2.2 map出来想要的数据结构
        val cidAndSidCount = cidSidAndCount.map { 
          case ((cid, sid), count) => (cid, (sid, count))
        }
        //2.3分组 排序取top10
        val cidAndSidCountItRDD: RDD[(Long, Iterable[(String, Int)])] = cidAndSidCount.groupByKey()
    
        //2.4 对每个值排序取top10 Iterable不能排序 要toList
        val result = cidAndSidCountItRDD.mapValues(it => { 
          //只能使用scala排序，Scala排序必须把所有数据加装内存才能排
          it.toList.sortBy(-_._2).take(10)
        })
    
        result.collect().foreach(println)
      }
        /* 解决方案2：每次排序一个cid，排10次 */
        def statCategorySessionTop10_2(sc:SparkContext,categoryTop10: Array[CategoryCountInfo],
          UserVisitActionRDD: RDD[UserVisitAction]) ={ 
          //1.过滤出来，只包含top10品类的那些点击记录
        //1.1先把top10品类id拿出来,转成long型的id 目的是为了和UserVisitActionRDD接轨
        val cids = categoryTop10.map(_.categoryId.toLong)
        val filteredUserVisitActionRDD = UserVisitActionRDD.filter(action => cids.contains(action.click_category_id))
    
        //2.需要排10次
          cids.foreach(cids=>{ 
            //2.1过滤出来点击id是cid的记录
            //filteredUserVisitActionRDD.filter(action=>cids == action.click_category_id)
            val cidUserVisitActionRDD = filteredUserVisitActionRDD.filter(cids == _.click_category_id)
            // 2.2 聚合
            val r = cidUserVisitActionRDD
              .map(action=>((action.click_category_id,action.session_id),1))
              .reduceByKey(_+_)
              .map{ 
                case ((cid,sid),count)=>(cid,(sid,count))
              }
              .sortBy(-_._2._2)
              .take(10)
                .groupBy(_._1)
                .map{ 
                  case (cid,arr) =>(cid,arr.map(_._2).toList)
                }
            println(r.toMap)
          })
    
      }
      /* 解决方案3： 找一个可以排序的集合，然后时刻保持集合中只有10最大的元素 */
      def statCategorySessionTop10_3(sc: SparkContext, categoryTop10: Array[CategoryCountInfo], UserVisitActionRDD: RDD[UserVisitAction]) = { 
        // 1.过滤出来只包含top10品类的那些点击记录
        // 1.1 先top10品类id拿出来, 转成Long id的目的是为了和UserVisitAction clickId兼容
        val cids: Array[Long] = categoryTop10.map(_.categoryId.toLong)
        val filteredUserVisitActionRDD: RDD[UserVisitAction] = UserVisitActionRDD.filter(action => cids.contains(action
          .click_category_id))
    
        // 2. 每个品类top10session的计算
        // 2.1 先map需要字段
        val cidSidAndOne: RDD[((Long, String), Int)] =
        filteredUserVisitActionRDD.map(action => ((action.click_category_id, action.session_id), 1))
        // 2.2 做聚合操作 得到 RDD[((cid, sid), count))]
        val cidSidAndCount: RDD[((Long, String), Int)] = cidSidAndOne.reduceByKey(_ + _)
        // 2.2 map出来想要的数据结构 RDD[(cid, (sid, count))]
        val cidAndSidCount: RDD[(Long, (String, Int))] = cidSidAndCount.map { 
          case ((cid, sid), count) => (cid, (sid, count))
        }
        // 2.3 分组 排序取top10
        val cidAndSidCountItRDD: RDD[(Long, Iterable[(String, Int)])] = cidAndSidCount.groupByKey()
    
        // 2.4 对每个value排序取top10
        val result = cidAndSidCountItRDD.mapValues((it: Iterable[(String, Int)]) => { 
          // 不要把Iterable直接转成list再排序.
          var set: mutable.TreeSet[SessionInfo] = mutable.TreeSet[SessionInfo]()
          it.foreach { 
            case (sid, count) =>
              val info = SessionInfo(sid, count)
              set += info
              if (set.size > 10) set = set.take(10)
          }
          set.toList
        })
    
        result.collect().foreach(println)
    
      }
      /* 解决方案4 */
      def statCategorySessionTop10_4(sc: SparkContext, categoryTop10: Array[CategoryCountInfo],
                                     UserVisitActionRDD: RDD[UserVisitAction]) = { 
        // 1.过滤出来只包含top10品类的那些点击记录
        // 1.1 先top10品类id拿出来, 转成Long id的目的是为了和UserVisitAction clickId兼容
        val cids: Array[Long] = categoryTop10.map(_.categoryId.toLong)
        val filteredUserVisitActionRDD: RDD[UserVisitAction] = UserVisitActionRDD.filter(action => cids.contains(action
          .click_category_id))
    
        // 2. 每个品类top10session的计算
        // 2.1 先map需要字段
        val cidSidAndOne: RDD[((Long, String), Int)] =
        filteredUserVisitActionRDD.map(action => ((action.click_category_id, action.session_id), 1))
        // 2.2 做聚合操作 得到 RDD[((cid, sid), count))]
        val cidSidAndCount: RDD[((Long, String), Int)] =
          cidSidAndOne.reduceByKey(new CategorySessionPartitioner(cids),_ + _)
    
        //2.3cidSidAndCount执行mapPartitions
        val result = cidSidAndCount.mapPartitions(it=>{ 
            // 不要把Iterable直接转成list再排序.
            var set: mutable.TreeSet[SessionInfo] = mutable.TreeSet[SessionInfo]()
          var categoryId = -1L
            it.foreach { 
              case ((cid,sid), count) =>
                categoryId = cid
                val info = SessionInfo(sid, count)
                set += info
                if (set.size > 10) set = set.take(10)
            }
          //set.map((categoryId,_)).toIterator
          Iterator((categoryId,set.toList))
    
        })
        result.collect().foreach(println)
    
      }
    
    }
    class CategorySessionPartitioner(cids: Array[Long]) extends Partitioner{ 
    
      private val cidIndexMap: Map[Long, Int] = cids.zipWithIndex.toMap
    
      //分区和品类id数量保持一致，可以保证一个分区一个cid
      override def numPartitions: Int = cids.length
    
      //(Long, String) => (cid,sessionId)
      override def getPartition(key: Any): Int = key match { 
          //使用这个cid在数组中的下标最为分区的索引非常合适
        case (cid:Long,_) =>cidIndexMap(cid)
      }
    }
    
    /* 需求一的方法步骤 1. 得到top10的品类的id 2. 过去出来只包含 top10 品类id的那些用户行为 3. 分组计算 => RDD[(品类id, sessionId))] map => RDD[(品类id, sessionId), 1)] reduceByKey => RDD[(品类id, sessionId), count)] map => RDD[品类id, (sessionId, count)] groupByKey RDD[品类id, Iterator[(sessionId, count)]] */

**`在bean里面创建一个Session类`**

case class SessionInfo(sessionId: String,
                           count: Long) extends Ordered[SessionInfo] { 
      // 按照降序排列
      // else if (this.count == that.count) 0 这个不能加, 否则会去重
      override def compare(that: SessionInfo): Int =
        if (this.count > that.count) -1
        else 1
    
    }

如果不懂排序和升序可以看这篇文章  
[https://blog.csdn.net/qq\_46548855/article/details/108176177][https_blog.csdn.net_qq_46548855_article_details_108176177]

## 完整项目代码 ##

**`记得在第一个需求的CategoryTopApp最后写result，返回结果，不要打印`**

import bean.UserVisitAction
    import org.apache.spark.rdd.RDD
    import org.apache.spark.{ SparkConf, SparkContext}
    
    object ProjectApp { 
      def main(args: Array[String]): Unit = { 
        val conf = new SparkConf().setMaster("local[*]").setAppName("project")
        val sc = new SparkContext(conf)
    
        //从数据把文件读出
        val sourceRDD = sc.textFile("D:\\idea\\spark-knight1\\input\\user_visit_action.txt")
    
        //把数据封装号(封装到样例类中)
        val userVisitActionRDD: RDD[UserVisitAction] = sourceRDD.map(line => { 
          val fields = line.split("_")
          UserVisitAction(
            fields(0),
            fields(1).toLong,
            fields(2),
            fields(3).toLong,
            fields(4),
            fields(5),
            fields(6).toLong,
            fields(7).toLong,
            fields(8),
            fields(9),
            fields(10),
            fields(11),
            fields(12).toLong)
        })
    
        //需求1：
        val categoryTop10 = CategoryTopApp.calcCategoryTop10(sc, userVisitActionRDD)
        //需求2：top10piping类的top10session
     CtegorySessionTopApp.statCategorySessionTop10_4(sc,categoryTop10,userVisitActionRDD)
    
      }
    
    }

## 3种方案的对比(建议用方案4) ##

问题解决方案：
        方案2：
           1.用spark排序，来解决问题
           2.spark的排序是整体排序，不能直接使用spark排序    例子：班级10个级  scala排是1个班前10  spark是10个班前10
           3.10个品类id，我就使用spark的排序功能排10次
    
          优点：
            一定能完成，不会oom
          缺点：
            起10个job，排序10次
    
         方案3：
            内存溢出 iterable=>转换list
    
    
            最终的目的top10
            搞一个集合，这个集合永远只保存10个元素，用于最大的10个元素
    
         方案4：
             对方案3做优化，减少一次shuffle
             减少shuffle只能去掉groupByKey
    
             还需要得到每个cid的所有session的集合？！  怎么得？
             rdd是分区得，mapPartitions(it=>{ })
             能不能让一个分区只有一个cid得所有数据
    
             每个分区只有一种cid,如何做到每个分区只有一个cid？
             用自定义分区器
             10cid，应该有10个分区

[https_github.com_lidonglin-bit_Spark-Core]: https://github.com/lidonglin-bit/Spark-Core/
[https_blog.csdn.net_qq_46548855_article_details_108198209]: https://blog.csdn.net/qq_46548855/article/details/108198209
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ2NTQ4ODU1_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/c03060525e6e452f907a971268caa0a6.png
[https_blog.csdn.net_qq_46548855_article_details_108176177]: https://blog.csdn.net/qq_46548855/article/details/108176177