Spark练习

第一个需求

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

代码：

取热门top10的方法：

def top10Category(rdd: RDD[String]): Array[String] = {
  rdd.flatMap(
    (action: String) => {
      val datas: Array[String] = action.split("_")
      if (datas(6) != "-1") {
        // 点击场合
        List((datas(6), (1, 0, 0)))
      } else if (datas(8) != "null") {
        // 下单场合
        datas(8).split(",").map((_, (0, 1, 0)))
      } else if (datas(10) != "null") {
        // 支付场合
        datas(10).split(",").map((_, (0, 0, 1)))
      } else {
        Nil
      }
    }
  ).reduceByKey(
    (t1, t2) => {
      (t1._1 + t2._1, t1._2 + t2._2, t1._3 + t2._3)
    }
  ).sortBy(_._2, false).take(10).map(_._1)
}

输出：

15
2
20
12
11
17
7
9
19
13

统计top10 session总计：

val value: RDD[String] = sparkContext.textFile("C:\\BaiduNetdiskDownload\\15.Spark\\2.资料\\spark-core数据\\user_visit_action.txt")
value.cache();
val strings: Array[String] = top10Category(value)
strings.toList.foreach(println)
//保留前10的数据
val value1: RDD[String] = value.filter((action: String) => {
  var datas = action.split("_")
  if (datas(6) != "-1") {
    strings.contains(datas(6))
  } else {
    false
  }
})
//    将商品(（id，session),sum）
val value2: RDD[((String, String), Int)] = value1.map(str => {
  val strings1 = str.split("_")
  ((strings1(6), strings1(2)), 1)
}).reduceByKey(_ + _)
//    将(（id，session),sum））=>(id,(session,sum))
val value3: RDD[(String, (String, Int))] = value2.map {
  case ((cid, session), sum) => (cid, (session, sum))
}
//分类
val groupdRdd: RDD[(String, Iterable[(String, Int)])] = value3.groupByKey()

val resultRDD: RDD[(String, List[(String, Int)])] = groupdRdd.mapValues(iters => {
  iters.toList.sortBy(_._2).reverse.take(10)

})
resultRDD.collect.foreach(println)