需求：每个热门品类 的 点击量 的 前十 用户

  1. object Spark05_Req2_HotCategoryTop10SessionAnalysis {
  3.     def main(args: Array[String]): Unit = {
  5.         // TODO : Top10热门品类
  6.         val sparConf = new SparkConf().setMaster("local[*]").setAppName("HotCategoryTop10Analysis")
  7.         val sc = new SparkContext(sparConf)
  9.         val actionRDD = sc.textFile("datas/user_visit_action.txt")
  11.         actionRDD.cache()
  13.         val top10Ids: Array[String] = top10Category(actionRDD)
  15.         // 1. 过滤原始数据,保留点击和前10品类ID
  16.         val filterActionRDD = actionRDD.filter(
  17.             action => {
  18.                 val datas = action.split("_")
  19.                 if ( datas(6) != "-1" ) {
  20.                     top10Ids.contains(datas(6))
  21.                 } else {
  22.                     false
  23.                 }
  24.             }
  25.         )
  27.         // 2. 根据品类ID和sessionid进行点击量的统计
  28.         val reduceRDD: RDD[((String, String), Int)] = filterActionRDD.map(
  29.             action => {
  30.                 val datas = action.split("_")
  31.                 ((datas(6), datas(2)), 1)
  32.             }
  33.         ).reduceByKey(_ + _)
  35.         // 3. 将统计的结果进行结构的转换
  36.         //  (（ 品类ID，sessionId ）,sum) => ( 品类ID，（sessionId, sum） )
  37.         val mapRDD = reduceRDD.map{
  38.             case ( (cid, sid), sum ) => {
  39.                 ( （cid）, (sid, sum) )
  40.             }
  41.         }
  43.         // 4. 相同的品类进行分组
  44.         val groupRDD: RDD[(String, Iterable[(String, Int)])] = mapRDD.groupByKey()
  46.         // 5. 将分组后的数据进行点击量的排序，取前10名
  47.         val resultRDD = groupRDD.mapValues(
  48.             iter => {
  49.                 iter.toList.sortBy(_._2)(Ordering.Int.reverse).take(10)
  50.             }
  51.         )
  53.         resultRDD.collect().foreach(println)
  55.         sc.stop()
  56.     }
  58.     def top10Category(actionRDD:RDD[String]) = {
  59.         val flatRDD: RDD[(String, (Int, Int, Int))] = actionRDD.flatMap(
  60.             action => {
  61.                 val datas = action.split("_")
  62.                 if (datas(6) != "-1") {
  63.                     // 点击的场合
  64.                     List((datas(6), (1, 0, 0)))
  65.                 } else if (datas(8) != "null") {
  66.                     // 下单的场合
  67.                     val ids = datas(8).split(",")
  68.                     ids.map(id => (id, (0, 1, 0)))
  69.                 } else if (datas(10) != "null") {
  70.                     // 支付的场合
  71.                     val ids = datas(10).split(",")
  72.                     ids.map(id => (id, (0, 0, 1)))
  73.                 } else {
  74.                     Nil
  75.                 }
  76.             }
  77.         )
  79.         val analysisRDD = flatRDD.reduceByKey(
  80.             (t1, t2) => {
  81.                 ( t1._1+t2._1, t1._2 + t2._2, t1._3 + t2._3 )
  82.             }
  83.         )
  85.         analysisRDD.sortBy(_._2, false).take(10).map(_._1)
  86.     }
  87. }