spark案例：Top10 热门品类

数据格式：

数据文件中每行数据采用下划线分隔数据

每一行数据表示用户的一次行为，这个行为只能是 4 种行为的一种

如果搜索关键字为 null,表示数据不是搜索数据

如果点击的品类 ID 和产品 ID 为-1，表示数据不是点击数据

针对于下单行为，一次可以下单多个商品，所以品类 ID 和产品 ID 可以是多个，id 之间采用逗号分隔，如果本次不是下单行为，则数据采用 null 表示

支付行为和下单行为类似

需求：按照每个品类的点击、下单、支付的量来统计热门品类（先按照点击数排名，靠前的就排名高；如果点击数相同，再比较下单数；下单数再相同，就比较支付数）

分析：数据可以统计成（品类，点击总数）（品类，下单总数）（品类，支付总数）这种格式来进行汇总，就简化成了WordCount问题

在实际的解决中我们可以进一步分析解决将数据转化成（品类，（点击，下单，支付））来解决

代码：

object demand {def main(args: Array[String]): Unit = {val sc = new SparkContext(new SparkConf().setMaster("local[*]").setAppName("demand"))val rdd: RDD[String] = sc.textFile("datas/user_visit_action.txt")// TODO 分别统计每个品类点击的次数，下单的次数和支付的次数： （品类，点击总数）（品类，下单总数）（品类，支付总数）val accumulator = new MyAccumulatorsc.register(accumulator,"demand")// 第一种 ：将数据扁平化整理成（品类，（点击，下单，支付）），按key值进行累加即可
//        rdd.flatMap(
//            value => {
//                val values: Array[String] = value.split("_")
//                if (values(6) != "-1") {
//                    List((values(6), (1, 0, 0)))
//                } else if (values(8) != "null") {
//                    val ids: Array[String] = values(8).split(",")
//                    ids.map(
//                        id => (id, (0, 1, 0))
//                    )
//                } else if (values(10) != "null") {
//                    val ids: Array[String] = values(10).split(",")
//                    ids.map(
//                        id => (id, (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).foreach(println)// 第二种：使用累加器来避免shufflerdd.foreach(value => {val values: Array[String] = value.split("_")if (values(6) != "-1") {accumulator.add(values(6),"click")} else if (values(8) != "null") {val ids: Array[String] = values(8).split(",")ids.foreach(id => accumulator.add(id,"order"))} else if (values(10) != "null") {val ids: Array[String] = values(10).split(",")ids.foreach(id => accumulator.add(id,"pay"))}})// 要自定义比较规则accumulator.value.map(_._2).toList.sortWith((l,r) =>{if (l.clickCnt > r.clickCnt){true}else if (l.clickCnt == r.clickCnt){if (l.orderCnt > r.orderCnt){true}else if (l.orderCnt == r.orderCnt){if (l.payCnt > r.payCnt){true}else false}else false}else false}).take(10).foreach(println)sc.stop()}// 辅助类case class UserVisitAction(cid:String,var clickCnt:Int,var orderCnt:Int,var payCnt:Int)// 自定义累加器class MyAccumulator extends AccumulatorV2[(String,String),mutable.Map[String, UserVisitAction]]{private var values: mutable.Map[String,UserVisitAction] = mutable.Map[String,UserVisitAction]()// 累加器是否为初始状态override def isZero: Boolean = values.isEmpty// 复制累加器override def copy(): AccumulatorV2[(String,String), mutable.Map[String,UserVisitAction]] = {new MyAccumulator()}// 重置累加器override def reset(): Unit = values.clear()// 向累加器中添加数据override def add(v: (String,String)): Unit = {val cid: String = v._1val action: String = v._2// 获取品类对应的UserVisitAction数据，从而来实现对相应操作的累加赋值val action1: UserVisitAction = values.getOrElse(cid, UserVisitAction(cid, 0, 0, 0))if (action == "click"){action1.clickCnt += 1}else if (action == "order"){action1.orderCnt += 1}else if (action == "pay"){action1.payCnt += 1}// 数据一定要进行更新values.update(cid,action1)}// 合并累加器override def merge(other: AccumulatorV2[(String,String), mutable.Map[String, UserVisitAction]]): Unit = {// 你将要输出的mapval map1 = this.values// 相对于输出的map以外的来自其他task任务返回的mapval map2 = other.value// 对结果进行merge合并map2.foreach{case (cid,use) => {val action: UserVisitAction = map1.getOrElse(cid, UserVisitAction(cid, 0, 0, 0))action.clickCnt += use.clickCntaction.orderCnt += use.orderCntaction.payCnt += use.payCnt// 一定要记得对数据更新到你要输出的map中去map1.update(cid,action)}}}// 返回累加器的结果override def value: mutable.Map[String, UserVisitAction] = values}
}

在解决问题时，由于spark有三种数据结构：RDD、累加器、广播变量，分别解决了一些问题，我们可以配合的使用它们来优化程序，由于计算时shuffle过程涉及磁盘IO，所有可以通过累加器的使用来避免shuffle提高效率。

spark案例：Top10 热门品类相关推荐

案例实操-Top10热门品类
需求1:分别统计每个品类的点击次数,下单次数和支付次数 (品类,点击总数)(品类,下单总数)(品类,支付总数) 排名顺序如:点击总数>下单总数>支付总数方案一 def main(args ...
50.Spark大型电商项目-用户访问session分析-top10热门品类之本地测试
本篇文章记录用户访问session分析-top10热门品类之本地测试. 在测试的过程中,到很多问题. 问题一:二次排序需要序列化,否则会在程序运行的时候报错. public class Categor ...
43.Spark大型电商项目-用户访问session分析-top10热门品类之需求回顾以及实现思路分析
目录需求回顾 top10热门品类二次排序实现思路分析本篇文章将记录用户访问session分析-top10热门品类之需求回顾以及实现思路分析. 需求回顾 top10热门品类计算出来通过筛选条件 ...
Spark Core项目实战(1) | 准备数据与计算Top10 热门品类(附完整项目代码及注释)
大家好,我是不温卜火,是一名计算机学院大数据专业大二的学生,昵称来源于成语-不温不火,本意是希望自己性情温和.作为一名互联网行业的小白,博主写博客一方面是为了记录自己的学习过程,另一方面是总结自己 ...
top10热门品类之需求以及实现思路分析
一. 需求:top10热门品类计算出来通过筛选条件的那些session,他们访问过的所有品类(点击.下单.支付),按照各个品类的点击.下单和支付次数,降序排序,获取前10个品类,也就是筛选条件下的那 ...
35.top10热门品类之需求回顾以及实现思路分析
本文为<Spark大型电商项目实战> 系列文章之一,主要介绍session访问分析里的top10热门品类模块,本文主要进行需求分析和实现思路. 需求回顾:top10热门品类计算出来通过筛 ...
SparkCore项目实战需求一Top10热门品类需求二Top10热门品类下每个品类的Top10活跃用户统计需求三计算页面单跳转换率
目录数据格式简介需求一:Top10热门品类(普通算子实现) 优化:需求一(使用ReduceByKey进行预聚合) 优化:需求一(采用累加器,避免shuffle过程) 需求二:Top10热门品类下每 ...
大数据之Spark案例实操完整使用(第六章)
大数据之Spark案例实操完整使用一.案例一 1.准备数据 2.需求 1:Top10 热门品类 3.需求说明方案一. 实现方案二实现方案三二 .需求实现 1.需求 2:Top10 热门品类中每 ...
Spark 案例（依据电商网站的真实需求）
目录数据说明需求1:Top10 热门品类需求说明实现方案一需求分析需求实现实现方案二需求分析需求实现实现方案三需求分析需求实现需求 2:Top10 热门品类中每个品类的 To ...

spark案例：Top10 热门品类

spark案例：Top10 热门品类相关推荐

最新文章

热门文章