第一章.RDD编程练习

1.需求一:每个省份点击量最多的三个广告

数据准备:时间戳,省份,城市,用户,广告,中间字段使用空格分隔

agent.log

  1. package com.atguigu.spark.demo
  3. import org.apache.spark.{SparkConf, SparkContext}
  5. /**
  6.  * 需求一:统计每个省份点击量最多的三个广告
  7.  * 数据准备:时间戳,省份,城市,用户,广告,中间字段使用空格分隔
  8.  */
  9. object demo1 {
  10.   def main(args: Array[String]): Unit = {
  11.     //创建SparkContext对象
  12.     val sc = new SparkContext(new SparkConf().setMaster("local[4]").setAppName("test"))
  13.     //读取数据
  14.     val datas = sc.textFile("datas/agent.log")
  15.     //列裁剪[广告,省份],去重,过滤
  16.     val selectRdd = datas.map(line => {
  17.       val arr = line.split(" ")
  18.       val province = arr(1)
  19.       val adid = arr.last
  20.       ((province, adid), 1)
  21.     })
  22.     //按照省份+广告分组
  23.     //统计每个省份,每个广告的点击次数
  24.     val provinceAdRdd = selectRdd.reduceByKey(_ + _)
  25.     //按照省份分组
  26.     val provinceRdd = provinceAdRdd.groupBy({
  27.       case ((province, adid), num) => province
  28.     })
  29.     //对每个省份的所有广告数据排序取前三
  30.     val top3Rdd = provinceRdd.map(x => {
  31.       val top3 = x._2.toList.sortBy({
  32.         case ((province, adid), num) => num
  33.       }).reverse
  34.         .take(3).map({
  35.         case ((province, adid), num) => (adid, num)
  36.       })
  37.       (x._1, top3)
  38.     })
  39.     //结果展示
  40.     top3Rdd.collect().toList.foreach(println(_))
  42.   }
  44. }

$11[Spark练习] - 图1

第二章.SparkCore实战

1.数据准备

user_visit_action.txt

$11[Spark练习] - 图2

2.需求一: Top10热门品类(第一种实现)

  • SQL分析
  1. select 点击品类id,count(1)
  2.    from T where 点击品类id!='-1'
  3.     group by 点击品类id
  4. left join
  5.   select 下单品类id,count(1)
  6.      from T where 下单品类id != 'null'
  7.      group by 下单品类id
  8. left join
  9.   select 支付品类id,count(1)
  10.      from T where 支付品类id != 'null'
  11.      group by 支付品类id
  • 代码实现
  1. package com.atguigu.spark.demo
  3. import org.apache.spark.{SparkConf, SparkContext}
  5. /**
  6.  * 获取top10热门品类(第一种实现)
  7.  * 根据品类id的点击数,下单数,支付数来确定
  8.  */
  9. object demo2 {
  10.   def main(args: Array[String]): Unit = {
  11.     val sc = new SparkContext(new SparkConf().setMaster("local[4]").setAppName("test"))
  13.     //1.读取数据
  14.     val source = sc.textFile("datas/user_visit_action.txt")
  15.     //2.过滤掉搜索数据
  16.     val filterSource = source.filter(line => line.split("_")(5) == "null")
  17.     //3.列裁剪[点击品类id,下单品类id,支付品类id]
  18.     val selectSource = filterSource.map(line => {
  19.       val arr = line.split("_")
  20.       //点击品类id
  21.       val clickid = arr(6)
  22.       //下单品类id
  23.       val orderids = arr(8)
  24.       //支付品类id
  25.       val payids = arr(10)
  26.       (clickid, orderids, payids)
  27.     })
  28.     //4.统计品类点击次数
  29.     //4.1过滤出点击数据
  30.     val clickRdd = selectSource.filter({
  31.       case (clickid, orderids, payids) => clickid != "-1"
  32.     })
  34.     //4.2统计出每个品类的点击次数
  35.     val clickNumRdd = clickRdd.map {
  36.       case (clickid, orderids, payids) => (clickid, 1)
  37.     }.reduceByKey(_ + _)
  39.     //5.统计品类下单次数
  40.     //5.1过滤出下单数据
  41.     val orderRdd = selectSource.filter({
  42.       case (clickid, orderids, payids) => orderids != "null"
  43.     })
  44.     //5.2切割下单品类id+压平
  45.     val orderExpload = orderRdd.flatMap({
  46.       case (clickid, orderids, payids) => {
  47.         val ss = orderids.split(",")
  48.         ss.map(id => (id, 1))
  49.       }
  50.     })
  53.     //5.3统计下单次数
  54.     val orderNumRdd = orderExpload.reduceByKey(_ + _)
  56.     //6.统计品类支付次数
  57.     //6.1过滤出支付数据
  58.     val payRdd = selectSource.filter({
  59.       case (clickid, orderids, payids) => payids != "null"
  60.     })
  61.     //6.2切割支付品类id+压平
  62.     val payExploadRdd = payRdd.flatMap({
  63.       case (clickid, orderids, payids) => {
  64.         payids.split(",").map(id => (id, 1))
  65.       }
  66.     })
  68.     //6.3统计支付次数
  69.     val payNumRdd = payExploadRdd.reduceByKey(_ + _)
  71.     //7.三者join得到每个品类的点击次数,下单次数,支付次数
  72.     val totalRdd = clickNumRdd.leftOuterJoin(orderNumRdd).leftOuterJoin(payNumRdd)
  74.     //8.按照点击次数,下单次数,支付次数排序取前10
  75.     val totalNumRdd = totalRdd.map({
  76.       case (id, ((clickNum, orderNum), paynum)) => (id, clickNum, orderNum.getOrElse(0), paynum.getOrElse(0))
  77.     })
  78.     val top10 = totalNumRdd.sortBy({
  79.       case (id, clickNum, orderNum, payNum) => (clickNum, orderNum, payNum)
  80.     }, false).take(10)
  82.     //9.结果展示
  83.     top10.foreach(println(_))
  84.   }
  85. }

$11[Spark练习] - 图3

共产生了4次shuffle,有5个stage,需要进行优化

$11[Spark练习] - 图4

3.需求一: Top10热门品类(第二种实现)

  • SQL分析

$11[Spark练习] - 图5

  • 代码实现
  1. package com.atguigu.spark.demo
  3. import org.apache.spark.rdd.RDD
  4. import org.apache.spark.{SparkConf, SparkContext}
  6. /**
  7.  * 获取top10热门品类(第二种实现方式)
  8.  */
  10. object demo3 {
  11.   def main(args: Array[String]): Unit = {
  12.     val sc = new SparkContext(new SparkConf().setMaster("local[4]").setAppName("test"))
  13.     //1.读取数据
  14.     val source = sc.textFile("datas/user_visit_action.txt")
  15.     //2.过滤(过滤掉搜索数据)
  16.     val filterSource = source.filter(line => {
  17.       line.split("_")(5) == "null"
  18.     })
  19.     //3.切割ids,识别行为+压平
  20.     val exploadRdd:RDD[(String,(Int,Int,Int))] = filterSource.flatMap(line => {
  21.       val arr = line.split("_")
  22.       val clickid = arr(6)
  23.       val orderids = arr(8)
  24.       val payids = arr(10)
  25.       //点击行为
  26.       if (clickid != "-1") {
  27.         (clickid, (1, 0, 0)) :: Nil
  28.       }
  29.       //下单行为
  30.       else if (orderids != "null") {
  31.         orderids.split(",").map(id => (id, (0, 1, 0)))
  32.       }
  33.       //支付行为
  34.       else {
  35.         payids.split(",").map(id => (id, (0, 0, 1)))
  36.       }
  37.     })
  38.     //4.分组聚合
  39.     val numRdd = exploadRdd.reduceByKey((agg, curr) => {
  40.       (agg._1 + curr._1, agg._2 + curr._2, agg._3 + curr._3)
  41.     })
  43.     //5.排序取前十
  44.     val top10 = numRdd.sortBy(_._2, false).take(10)
  46.     //6.结果展示
  47.     top10.foreach(println(_))
  48.     Thread.sleep(1000000)
  49.   }
  51. }

$11[Spark练习] - 图6

共产生了两次shuffle,有3个stage

$11[Spark练习] - 图7

4.需求一: Top10热门品类(第三种实现)

使用累加器代替第二种实现方式的reduceByKey,其余代码不变

  • 自定义累加器
  1. package com.atguigu.spark.demo
  3. import org.apache.spark.util.AccumulatorV2
  5. import scala.collection.mutable
  7. class Top10Accumulator extends AccumulatorV2[(String,(Int,Int,Int)),mutable.Map[String,(Int,Int,Int)]]{
  8.   //创建一个中间结果容器
  9.   val map = mutable.Map[String, (Int, Int, Int)]()
  10.   //判断累加器是否为空
  11.   override def isZero: Boolean = map.isEmpty
  12.   //复制新的累加器
  13.   override def copy(): AccumulatorV2[(String, (Int, Int, Int)), mutable.Map[String, (Int, Int, Int)]] = new Top10Accumulator
  14.   //重置累加器
  15.   override def reset(): Unit = map.clear()
  16.   //在每个分区中累加元素
  17.   override def add(v: (String, (Int, Int, Int))): Unit = {
  18.     //获取当前品类之前的统计次数
  19.     val beforeCount = map.getOrElse(v._1, (0, 0, 0))
  20.     //统计总次数
  21.     val totalCount = (beforeCount._1 + v._2._1, beforeCount._2 + v._2._2, beforeCount._3 + v._2._3)
  22.     map.put(v._1,totalCount)
  23.   }
  24.   //在Driver中对task结果汇总
  25.   override def merge(other: AccumulatorV2[(String, (Int, Int, Int)), mutable.Map[String, (Int, Int, Int)]]): Unit = {
  26.     //获取task的累加结果
  27.     val taskResult = other.value
  28.     //将Driver的中间结果与当前task统计结果合并
  29.     val totalList = taskResult.toList ++ map.toList
  30.     //根据品类id分组
  31.     val groupedMap = totalList.groupBy(_._1)
  32.     //统计每个品类的总次数
  33.     val result = groupedMap.map(x => {
  34.       x._2.reduce((agg, curr) => {
  35.         (agg._1, (agg._2._1 + curr._2._1, agg._2._2 + curr._2._2, agg._2._3 + curr._2._3))
  36.       })
  37.     })
  38.     map.++=(result)
  39.   }
  41.   override def value: mutable.Map[String, (Int, Int, Int)] = map
  42. }
  • 代码实现
  1. package com.atguigu.spark.demo
  3. import org.apache.spark.{SparkConf, SparkContext}
  4. import org.apache.spark.rdd.RDD
  6. /**
  7.  * 获取top10热门品类(第三种实现方式)
  8.  * 使用累加器
  9.  */
  10. object demo4 {
  11.   def main(args: Array[String]): Unit = {
  12.     val sc = new SparkContext(new SparkConf().setMaster("local[4]").setAppName("test"))
  13.     //注册累加器
  14.     val acc = new Top10Accumulator
  15.     sc.register(acc,"top10")
  16.     //1.读取数据
  17.     val source = sc.textFile("datas/user_visit_action.txt")
  18.     //2.过滤(过滤掉搜索数据)
  19.     val filterSource = source.filter(line => {
  20.       line.split("_")(5) == "null"
  21.     })
  22.     //3.切割ids,识别行为+压平
  23.     val exploadRdd:RDD[(String,(Int,Int,Int))] = filterSource.flatMap(line => {
  24.       val arr = line.split("_")
  25.       val clickid = arr(6)
  26.       val orderids = arr(8)
  27.       val payids = arr(10)
  28.       //点击行为
  29.       if (clickid != "-1") {
  30.         (clickid, (1, 0, 0)) :: Nil
  31.       }
  32.       //下单行为
  33.       else if (orderids != "null") {
  34.         orderids.split(",").map(id => (id, (0, 1, 0)))
  35.       }
  36.       //支付行为
  37.       else {
  38.         payids.split(",").map(id => (id, (0, 0, 1)))
  39.       }
  40.     })
  41.     exploadRdd.foreach(x=>acc.add(x))
  42.     //获取所有品类的累加结果
  43.     val map = acc.value
  44.     map.toList.sortBy(_._2).reverse.take(10).foreach(println(_))
  47.     Thread.sleep(1000000)
  48.   }
  50. }

$11[Spark练习] - 图8

没有产生一次shuffle

$11[Spark练习] - 图9

5.需求二: Top10热门品类中每个品类的Top10活跃Session统计

$11[Spark练习] - 图10

  1. package com.atguigu.spark.demo
  3. import org.apache.spark.{SparkConf, SparkContext}
  4. import org.apache.spark.rdd.RDD
  5. /**
  6.  * Top10热门品类中每个品类的Top10活跃Session统计
  7.  *
  8.  */
  9. object demo5 {
  10.   def main(args: Array[String]): Unit = {
  11.     val sc = new SparkContext(new SparkConf().setMaster("local[4]").setAppName("test"))
  12.     //1.获取top10热门品类id
  13.     //注册累加器
  14.     val acc = new Top10Accumulator
  15.     sc.register(acc,"top10")
  16.     //读取数据
  17.     val source = sc.textFile("datas/user_visit_action.txt")
  18.     //过滤(过滤掉搜索数据)
  19.     val filterSource = source.filter(line => {
  20.       line.split("_")(5) == "null"
  21.     })
  22.     //切割ids,识别行为+压平
  23.     val exploadRdd:RDD[(String,(Int,Int,Int))] = filterSource.flatMap(line => {
  24.       val arr = line.split("_")
  25.       val clickid = arr(6)
  26.       val orderids = arr(8)
  27.       val payids = arr(10)
  28.       //点击行为
  29.       if (clickid != "-1") {
  30.         (clickid, (1, 0, 0)) :: Nil
  31.       }
  32.       //下单行为
  33.       else if (orderids != "null") {
  34.         orderids.split(",").map(id => (id, (0, 1, 0)))
  35.       }
  36.       //支付行为
  37.       else {
  38.         payids.split(",").map(id => (id, (0, 0, 1)))
  39.       }
  40.     })
  41.     exploadRdd.foreach(x=>acc.add(x))
  42.     //获取所有品类的累加结果
  43.     val map = acc.value
  44.     val top10List = map.toList.sortBy(_._2).reverse.take(10)
  45.     val top10ids = top10List.map(_._1)
  47.     //广播热门品类id
  48.     val bc = sc.broadcast(top10ids)
  50.     //2.过滤[只保留热门品类数据,只需要点击数据]
  51.     val filterRdd = source.filter(line => {
  52.       val arr = line.split("_")
  53.       val clickid = arr(6)
  54.       clickid != "-1" && bc.value.contains(clickid)
  55.     })
  56.     //3.列裁剪[点击的品类id,sessionid]
  57.     val sessionRdd = filterRdd.map(line => {
  58.       val arr = line.split("_")
  59.       ((arr(6), arr(2)), 1)
  60.     })
  61.     //4.统计每个热门品类 每个session的次数
  62.     val sessionClickRdd = sessionRdd.reduceByKey(_+_)
  63.     //5.按照热门品类id分组
  64.     val classRdd = sessionClickRdd.groupBy({
  65.       case ((id, sessionid), num) => id
  66.     })
  67.     //6.对每个品类的所有session数据排序取前十
  68.     val resRdd = classRdd.map(x => {
  69.       //获取前十session
  70.       val top10Session = x._2.toList.sortBy(_._2).reverse.take(10)
  71.       val top10 = top10Session.map( {
  72.         case ((id, session), num) => (session, num)
  73.       })
  74.       (x._1, top10)
  75.     })
  76.     //7.结果展示
  77.     resRdd.collect().foreach(println(_))
  78.   }
  79. }

$11[Spark练习] - 图11

6.需求三: 页面单跳转换率统计

  1. package com.atguigu.spark.demo
  3. import org.apache.spark.{SparkConf, SparkContext}
  5. /**
  6.  * 页面单跳转化率统计
  7.  */
  8. object demo6 {
  9.   def main(args: Array[String]): Unit = {
  10.     val list = List(1,2,3,4,5,6,7)
  11.     //获取待求取页面的跳转数据
  12.     val pagelist = list.init.zip(list.tail)
  13.     val sc = new SparkContext(new SparkConf().setMaster("local[4]").setAppName("test"))
  14.     //1.读取数据
  15.     val source = sc.textFile("datas/user_visit_action.txt")
  16.     //2.过滤数据(只要点击数据)
  17.     val clickRdd = source.filter(line => line.split("_")(6) != "-1")
  18.     //3.列裁剪[页面id,session,时间]
  19.     val selectRdd = clickRdd.map(line => {
  20.       val arr = line.split("_")
  21.       val userid = arr(1)
  22.       val pageid = arr(3).toInt
  23.       val sessionid = arr(2)
  24.       val time = arr(4)
  25.       (userid, sessionid, pageid, time)
  27.     })
  28.     //4.获取分母
  29.     //4.1过滤出1,2,3,4,5,6页面的数据
  30.     val fmPageRdd = selectRdd.filter {
  31.       case (userid, sessionid, pageid, time) => list.init.contains(pageid)
  32.     }
  33.     //4.2转换数据类型:页面id->1
  34.     val fmMapRdd = fmPageRdd.map(x => (x._3, 1))
  35.     //4.3统计页面访问总次数
  36.     val fmMap = fmMapRdd.reduceByKey(_ + _).collect().toMap
  38.     //5.获取分子[1->2]
  39.     //5.1按照用户和session分组
  40.     val sessionRdd = selectRdd.groupBy({
  41.       case (userid, sessionid, pageid, time) => (userid, sessionid)
  42.     })
  43.     val fromToPageRdd = sessionRdd.flatMap(x => {
  44.       //5.2对每个用户每个session的数据按照时间进行排序(升序)
  45.       val sortedList = x._2.toList.sortBy(_._4)
  46.       //5.3.滑窗(得到从哪个页面跳到哪个页面)
  47.       val slidingList = sortedList.sliding(2)
  48.       val fromToPageList = slidingList.map(x => {
  49.         val fromPage = x.head._3
  50.         val toPage = x.last._3
  51.         ((fromPage, toPage), 1)
  52.       })
  53.       //5.4.过滤掉不需要的跳转数据
  54.       fromToPageList.filter(x => {
  55.         pagelist.contains(x._1)
  56.       })
  57.     })
  58.     //5.6统计页面跳转的总次数
  59.     val fzMap = fromToPageRdd.reduceByKey(_ + _).collect().toMap
  60.     //求得转化率
  61.     pagelist.foreach({
  62.       case(fromPage,toPage)=>
  63.         //获取fromPage页面访问总次数
  64.         val fm = fmMap.getOrElse(fromPage, 1)
  65.         val fz = fzMap.getOrElse((fromPage,toPage),0)
  66.         println(s"从${fromPage}跳到${toPage}的转化率=${(fz.toDouble*100)/fm}%")
  67.     })
  68.   }
  70. }

$11[Spark练习] - 图12

7.需求四: 获取用户每次会话的行为轨迹

  1. - 数据
  2. "1001","2020-09-10 10:21:21","home.html",
  3. "1001","2020-09-10 10:28:10","good_list.html",
  4. "1001","2020-09-10 10:35:05","good_detail.html",
  5. "1001","2020-09-10 10:42:55","cart.html",
  6. "1001","2020-09-10 11:35:21","home.html",
  7. "1001","2020-09-10 11:36:10","cart.html",
  8. "1001","2020-09-10 11:38:12","trade.html",
  9. "1001","2020-09-10 11:40:00","payment.html",
  10. "1002","2020-09-10 09:40:00","home.html",
  11. "1002","2020-09-10 09:41:00","mine.html",
  12. "1002","2020-09-10 09:42:00","favor.html",
  13. "1003","2020-09-10 13:10:00","home.html",
  14. "1003","2020-09-10 13:15:00","search.html"
  1. package com.atguigu.spark.demo
  3. import org.apache.spark.{SparkConf, SparkContext}
  5. import java.text.SimpleDateFormat
  6. import java.util.UUID
  8. /**
  9.  * 需求:获取用户每次会话的行为轨迹
  10.  */
  11. object demo7 {
  12.   def main(args: Array[String]): Unit = {
  13.     val list = List[(String,String,String)](
  14.       ("1001","2020-09-10 10:21:21","home.html"),
  15.       ("1001","2020-09-10 10:28:10","good_list.html"),
  16.       ("1001","2020-09-10 10:35:05","good_detail.html"),
  17.       ("1001","2020-09-10 10:42:55","cart.html"),
  18.       ("1001","2020-09-10 11:35:21","home.html"),
  19.       ("1001","2020-09-10 11:36:10","cart.html"),
  20.       ("1001","2020-09-10 11:38:12","trade.html"),
  21.       ("1001","2020-09-10 11:40:00","payment.html"),
  22.       ("1002","2020-09-10 09:40:00","home.html"),
  23.       ("1002","2020-09-10 09:41:00","mine.html"),
  24.       ("1002","2020-09-10 09:42:00","favor.html"),
  25.       ("1003","2020-09-10 13:10:00","home.html"),
  26.       ("1003","2020-09-10 13:15:00","search.html")
  27.     )
  28.     val sc = new SparkContext(new SparkConf().setMaster("local[4]").setAppName("test"))
  29.     val rdd = sc.parallelize(list)
  30.     //1.数据类型转换
  31.     val rdd2 = rdd.map {
  32.       case (userid, timestr, page) =>
  33.         val formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
  34.         val time = formatter.parse(timestr).getTime
  35.         UserAnalysis(userid, time, page)
  36.     }
  37.     //rdd2.foreach(println(_))
  38.     //2.按照用户分组
  39.     val rdd3 = rdd2.groupBy(_.userid)
  40.     //3.对每个用户的所有数据按照时间排序
  41.     val rdd4 = rdd3.flatMap(x => {
  42.       //4.滑窗[数据两两比较,如果时间小于等于半小时,属于一次会话,时间大于半小时,属于不同会话]
  43.       val slidingList = x._2.toList.sortBy(_.time).sliding(2)
  44.       slidingList.foreach(list => {
  45.         val headElement = list.head
  46.         val lastElement = list.last
  47.         if (lastElement.time - headElement.time <= 30 * 60 * 1000) {
  48.           lastElement.session = headElement.session
  49.           lastElement.step = headElement.step + 1
  50.         }
  51.       })
  52.       x._2
  53.     })
  54.     //5.结果展示
  55.     rdd4.collect().foreach(println(_))
  57.   }
  58. }
  59. case class UserAnalysis(userid:String,time:Long,page:String,var session:String=UUID.randomUUID().toString,var step:Int=1)

$11[Spark练习] - 图13