• Spark案例实际操作

    目录

    数据说明

    需求1：Top10热门品类

    1.1 需求说明

    1.2 实现方案一

    1.2.1 需求分析

    1.2.2 需求实现

    1.2.3 问题

    1.3 实现方案二

    1.3.1 需求分析

    1.3.2 需求实现---结局方案一之中的问题

    1.3.3 问题

    1.4 实现方案三

    1.4.1 需求分析

    1.4.2 需求实现

    1.5 实现方案四

    1.5.1 需求分析

    1.5.2 需求实现

    需求2：Top10热门品类中每个品类的Top10活跃Session统计

    2.1 需求说明

    2.2需求分析

    2.3 功能实现(这里代码之中的架构需要进行注意)

    HotCataegoryTop10的代码 

    HotCategoryTop10SessionController

    需求3：页面单跳转换率统计

    3.1 需求说明

    3.2 需求分析

    数据说明

    • 上面的黑色底框的数据对应下面的白色底框之中的数据,只是分隔符是不一样的

    说明:上面的数据图是从数据文件中截取的一部分内容，表示为电商网站的用户行为数据，主要包含用户的4种行为：搜索，点击，下单，支付。数据规则如下：

    • 数据文件中每行数据采用下划线分隔数据
    • 每一行数据表示用户的一次行为，这个行为只能是4种行为的一种
    • 如果搜索关键字为null,表示数据不是搜索数据
    • 如果点击的品类ID和产品ID为-1，表示数据不是点击数据
    • 针对于下单行为，一次可以下单多个商品，所以品类ID和产品ID可以是多个，id之间采用逗号分隔，如果本次不是下单行为，则数据采用null表示
    • 支付行为和下单行为类似

    详细字段说明:

    样品类:

    1. //用户访问动作表
    2. case class UserVisitAction(
    3.     date: String,//用户点击行为的日期
    4.     user_id: Long,//用户的ID
    5.     session_id: String,//Session的ID
    6.     page_id: Long,//某个页面的ID
    7.     action_time: String,//动作的时间点
    8.     search_keyword: String,//用户搜索的关键词
    9.     click_category_id: Long,//某一个商品品类的ID
    10.     click_product_id: Long,//某一个商品的ID
    11.     order_category_ids: String,//一次订单中所有品类的ID集合
    12.     order_product_ids: String,//一次订单中所有商品的ID集合
    13.     pay_category_ids: String,//一次支付中所有品类的ID集合
    14.     pay_product_ids: String,//一次支付中所有商品的ID集合
    15.     city_id: Long
    16. )//城市 id

    需求1：Top10热门品类

    1.1 需求说明

    品类是指产品的分类，大型电商网站品类分多级，咱们的项目中品类只有一级，不同的公司可能对热门的定义不一样。我们按照每个品类的点击、下单、支付的量来统计热门品类。

    鞋 点击数 下单数  支付数

    衣服 点击数 下单数  支付数

    电脑 点击数 下单数  支付数

    例如，综合排名 = 点击数*20%+下单数*30%+支付数*50%

    本项目需求优化为：先按照点击数排名，靠前的就排名高；如果点击数相同，再比较下单数；下单数再相同，就比较支付数。

    1.2 实现方案一

    1.2.1 需求分析

    分别统计每个品类点击的次数，下单的次数和支付的次数：

    （品类，点击总数）（品类，下单总数）（品类，支付总数）

    1.2.2 需求实现

    • 奇怪的面试题，两个杯子，分别是五升水和三升水的杯子，如何得到四升水？？

    方式一：5+5-(3+3)=4.首先是将五升水的杯子之中的水倒入三升水空杯子之中，倒出三升水杯子之中的水，将五升水杯子之中剩余的两升水倒入三升水杯子之中。灌满五升水的杯子，将五升水的杯子倒入只有两升水的另一个杯子，剩下的就是四升水。

    方式二:3+3+3-5=4,首先将三升水倒入空的五升水的水杯,在将三升水灌满水倒入五升水的水杯,此刻三升水的水杯剩下一升水,将五升水的水杯之中的水倒掉,三升水的水杯之中的水转移到五生水的水杯之中,再将三升水导入五升水的水杯之中,得到四升水.

    • 补充:如何自动调出RDD的类型,设置步骤是如下所示:
    • 需求实现代码: 
      1. package req
      2.  
      3. import org.apache.spark.rdd.RDD
      4. import org.apache.spark.{SparkConf, SparkContext}
      5.  
      6. /**
      7.   * @anthor Yang
      8.   * @create 2022-07-21 22:47
      9.   */
      10. object Spark01_Req_HotCategoryTop10 {
      11.   def main(args: Array[String]): Unit = {
      12.     val conf= new SparkConf().setMaster("local[*]").setAppName("HotCategoryTop10")
      13.     val sc = new SparkContext(conf)
      14.  
      15.     //TODO 需求一：Top10热门品类
      16.     //TODO 获取原始文件
      17.     val fileDatas = sc.textFile("data/user_visit_action.txt")
      18.  
      19.     //TODO 统计分析之前需要将不需要的数据进行过滤-先保留所有的统计数据-对点击数据进行统计
      20.     val clickDatas = fileDatas.filter(
      21.       data => {
      22.         val datas = data.split("_")
      23.         val cid = datas(6)
      24.         cid != "-1"
      25.       }
      26.     )
      27.     //TODO 统计品类点击数量
      28.     val clickCntDatas = clickDatas.map(
      29.       data => {
      30.         val datas = data.split("_")
      31.         val cid = datas(6)
      32.         (cid,1)
      33.       }
      34.     ).reduceByKey(_+_)
      35.     //TODO 统计品类下单数量(方法是和上面差不多的)
      36.     val orderDatas = fileDatas.filter(
      37.       data => {
      38.         val datas = data.split("_")
      39.         val cid = datas(8)
      40.         cid != "null"
      41.       }
      42.     )
      43.     //(1,2,3,4)->(1,1) (2,1) (3.1) 整体变成个体是使用扁平化
      44.     //TODO 整体变成部分就是进行一个扁平化的过程
      45.     val orderCntDatas = orderDatas.flatMap(
      46.       data => {
      47.         val datas = data.split("_")
      48.         val cid = datas(8)
      49.         val cids = cid.split(",")
      50.         cids.map((_,1))
      51.       }
      52.     ).reduceByKey(_+_)
      53.  
      54.     //TODO 统计品类支付数量
      55.     val payDatas = fileDatas.filter(
      56.       data => {
      57.         val datas = data.split("_")
      58.         val cid = datas(10)
      59.         cid != "null"
      60.       }
      61.     )
      62.     //(1,2,3,4)->(1,1) (2,1) (3.1) 整体变成个体是使用扁平化
      63.     //TODO 整体变成部分就是进行一个扁平化的过程
      64.     val payCntDatas = payDatas.flatMap(
      65.       data => {
      66.         val datas = data.split("_")
      67.         val cid = datas(10)
      68.         val cids = cid.split(",")
      69.         cids.map((_,1))
      70.       }
      71.     ).reduceByKey(_+_)
      72.  
      73.     //TODO 对统计结果进行排序 =》 点击，下单，支付（三个指标全部进行排序完成之后进行选取 -> Tuple）
      74.     //val clickSortedDatas = clickCntDatas.sortBy(_._2,false) //这里不是用sortByKey，sortByKet只是对于Key进行一个排序的过程,false表示是降序的过程
      75.  
      76.     //TODO 将点击-》Tuple（品类ID，点击）
      77.     //TODO 下单-》Tuple（品类ID，下单）
      78.     //TODO 支付-》Tuple（品类ID，支付）
      79.     //TODO 将上述的数据变成（品类ID，（点击，下单，支付））
      80.     //上述的方式是不能采用JOIN的方式，因为有的数据可能是空的，join要求两边都是有数据的，join的底层是cogroup
      81.      //这里不使用下面这一条语句的原因是由于括号的原因
      82.     // val value: RDD[(String, (Option[(Option[Int], Option[Int])], Option[Int]))] = clickCntDatas.fullOuterJoin(orderCntDatas).fullOuterJoin(payCntDatas)
      83.     val value: RDD[(String, (Iterable[Int], Iterable[Int], Iterable[Int]))] = clickCntDatas.cogroup(orderCntDatas,payCntDatas)
      84.  
      85.     val mapDatas = value.map{
      86.       case(cid,(clickIter,orderIter,payIter)) =>
      87.         {
      88.           var clickcnt =0
      89.           var ordercnt =0
      90.           var paycnt = 0
      91.           val iterator1: Iterator[Int] = clickIter.iterator
      92.           if(iterator1.hasNext)
      93.             {
      94.               clickcnt = iterator1.next()
      95.             }
      96.           val iterator2: Iterator[Int] = orderIter.iterator
      97.           if(iterator2.hasNext)
      98.           {
      99.             ordercnt = iterator2.next()
      100.           }
      101.           val iterator3: Iterator[Int] = payIter.iterator
      102.           if(iterator3.hasNext)
      103.           {
      104.             paycnt = iterator3.next()
      105.           }
      106.           (cid,(clickcnt,ordercnt,paycnt))
      107.         }
      108.     }
      109.     
      110.       val top10: Array[(String, (Int, Int, Int))] = mapDatas.sortBy(_._2,false).take(10)
      111.  
      112.     //TODO 将统计采集之后的结果打印到控制台上
      113.     top10.foreach(println)
      114.  
      115.     sc.stop()
      116.   }
      117.  
      118. }

    1.2.3 问题

    1.三个过滤,同一个RDD(过滤)的重复使用.RDD是不保留数据的,重复使用浪费性能.

    2.cogroup可能存在笛卡尔乘积的问题,相同的时候出现OneToOneDependency,不相同的时候可能会出现ShuffleDependency的现象.------性能瓶颈(可能会出现问题),十几万数据如果要是出现笛卡尔乘积,算子性能会低下.

    1.3 实现方案二

    1.3.1 需求分析

    一次性统计每个品类点击的次数，下单的次数和支付的次数：

    （品类，（点击总数，下单总数，支付总数））

    1.3.2 需求实现---结局方案一之中的问题

    采用reduceByKey的方式代替cogroup,避免出现笛卡尔乘积的问题.

    1. package req
    2.  
    3. import org.apache.spark.rdd.RDD
    4. import org.apache.spark.{SparkConf, SparkContext}
    5.  
    6. /**
    7.   * @anthor Yang
    8.   * @create 2022-07-21 22:47
    9.   */
    10. object Spark02_Req_HotCategoryTop10 {
    11.   def main(args: Array[String]): Unit = {
    12.     val conf= new SparkConf().setMaster("local[*]").setAppName("HotCategoryTop10")
    13.     val sc = new SparkContext(conf)
    14.  
    15.     //TODO 需求一：Top10热门品类
    16.     //TODO 获取原始文件
    17.     val fileDatas = sc.textFile("data/user_visit_action.txt")
    18.  
    19.     //TODO 统计分析之前需要将不需要的数据进行过滤-先保留所有的统计数据-对点击数据进行统计
    20.     val clickDatas = fileDatas.filter(
    21.       data => {
    22.         val datas = data.split("_")
    23.         val cid = datas(6)
    24.         cid != "-1"
    25.       }
    26.     )
    27.  
    28.     //TODO 统计品类点击数量
    29.     val clickCntDatas = clickDatas.map(
    30.       data => {
    31.         val datas = data.split("_")
    32.         val cid = datas(6)
    33.         (cid,1)
    34.       }
    35.     ).reduceByKey(_+_)
    36.     //TODO 统计品类下单数量(方法是和上面差不多的)
    37.     val orderDatas = fileDatas.filter(
    38.       data => {
    39.         val datas = data.split("_")
    40.         val cid = datas(8)
    41.         cid != "null"
    42.       }
    43.     )
    44.     //(1,2,3,4)->(1,1) (2,1) (3.1) 整体变成个体是使用扁平化
    45.     //TODO 整体变成部分就是进行一个扁平化的过程
    46.     val orderCntDatas = orderDatas.flatMap(
    47.       data => {
    48.         val datas = data.split("_")
    49.         val cid = datas(8)
    50.         val cids = cid.split(",")
    51.         cids.map((_,1))
    52.       }
    53.     ).reduceByKey(_+_)
    54.  
    55.     //TODO 统计品类支付数量
    56.     val payDatas = fileDatas.filter(
    57.       data => {
    58.         val datas = data.split("_")
    59.         val cid = datas(10)
    60.         cid != "null"
    61.       }
    62.     )
    63.     //(1,2,3,4)->(1,1) (2,1) (3.1) 整体变成个体是使用扁平化
    64.     //TODO 整体变成部分就是进行一个扁平化的过程
    65.     val payCntDatas = payDatas.flatMap(
    66.       data => {
    67.         val datas = data.split("_")
    68.         val cid = datas(10)
    69.         val cids = cid.split(",")
    70.         cids.map((_,1))
    71.       }
    72.     ).reduceByKey(_+_)
    73.  
    74.     //TODO 对统计结果进行排序 =》 点击，下单，支付（三个指标全部进行排序完成之后进行选取 -> Tuple）
    75.     //val clickSortedDatas = clickCntDatas.sortBy(_._2,false) //这里不是用sortByKey，sortByKet只是对于Key进行一个排序的过程,false表示是降序的过程
    76.  
    77.     //TODO 将点击-》Tuple（品类ID，点击）  ---(品类ID,(点击,0,0))
    78.     //TODO 下单-》Tuple（品类ID，下单）   ---(品类ID,(0,下单,0))
    79.     //TODO 支付-》Tuple（品类ID，支付）   ---(品类ID,(0,0,支付))
    80.     //TODO 将上述的数据变成（品类ID，（点击，下单，支付））
    81.     //上述的方式是不能采用JOIN的方式，因为有的数据可能是空的，join要求两边都是有数据的，join的底层是cogroup
    82.      //这里不使用下面这一条语句的原因是由于括号的原因
    83.     // val value: RDD[(String, (Option[(Option[Int], Option[Int])], Option[Int]))] = clickCntDatas.fullOuterJoin(orderCntDatas).fullOuterJoin(payCntDatas)
    84.     //TODO reduceByKey存在shuffle,但是不存在笛卡尔乘积----修改的地方
    85.     val clickMapDatas = clickCntDatas.map{
    86.       case( cid,clickCnt ) => {
    87.         (cid , (clickCnt,0,0))
    88.       }
    89.     }
    90.     val orderkMapDatas = orderCntDatas.map{
    91.       case( cid,orderCnt ) => {
    92.         (cid , (0,orderCnt,0))
    93.       }
    94.     }
    95.     val payMapDatas = payCntDatas.map{
    96.       case( cid,payCnt ) => {
    97.         (cid , (0,0,payCnt))
    98.       }
    99.     }
    100.     //将上述的三个数据进行合并,使用union
    101.     val unionRDD: RDD[(String, (Int, Int, Int))] = clickMapDatas.union(orderkMapDatas).union(payMapDatas)
    102.     //使用reduceByKey
    103.     val reduceRDD = unionRDD.reduceByKey(
    104.       (t1,t2) =>
    105.         {
    106.           (t1._1+t2._1 ,t1._2+t2._2 ,t1._3+t2._3)
    107.         }
    108.     )
    109.     val top10 = reduceRDD.sortBy(_._2,false).take(10)
    110.  
    111.  
    112.     //TODO 将统计采集之后的结果打印到控制台上
    113.     top10.foreach(println)
    114.  
    115.     sc.stop()
    116.   }
    117.  
    118. }

    1.3.3 问题

    reduceByKey存在shuffle的问题,这里我们使用了四个reduceByKey,使用的shuffle次数太多,那么落盘的次数就是很多,想要提高性能,就要想办法减少落盘的过程.

    1.4 实现方案三

    1.4.1 需求分析

    对于上述的问题再次改进

    1.4.2 需求实现

    • 没有使用累加器,但是减少了shuffle.过滤的过程 
      1. package req
      2.  
      3. import org.apache.spark.rdd.RDD
      4. import org.apache.spark.{SparkConf, SparkContext}
      5. import org.codehaus.jackson.annotate.JsonTypeInfo.Id
      6.  
      7. /**
      8.   * @anthor Yang
      9.   * @create 2022-07-21 22:47
      10.   */
      11. object Spark03_Req_HotCategoryTop10 {
      12.   def main(args: Array[String]): Unit = {
      13.     val conf= new SparkConf().setMaster("local[*]").setAppName("HotCategoryTop10")
      14.     val sc = new SparkContext(conf)
      15.  
      16.     //TODO 需求一：Top10热门品类
      17.     //TODO 获取原始文件
      18.     val fileDatas = sc.textFile("data/user_visit_action.txt")
      19.  
      20.     val flatDatas = fileDatas.flatMap(//flatMap要求返回的是可以迭代的集合
      21.       data => {
      22.         var datas = data.split("_")
      23.         if( datas(6) != "-1" )
      24.           {//点击数据的场合
      25.             List((datas(6),(1,0,0)))
      26.           }
      27.         else if(datas(8) != "null")
      28.           {//下单的场合
      29.             val id = datas(8)
      30.            val ids = id.split(",")
      31.             ids.map(//map返回就是迭代的集合
      32.               id => {
      33.                (id,(0,1,0))
      34.               }
      35.             )
      36.  
      37.           }
      38.         else if(datas(10) != "null")
      39.           {
      40.             val id= datas(10)
      41.            val ids = id.split(",")
      42.             ids.map(
      43.               id=> {
      44.                (id,(0,0,1))
      45.               }
      46.             )
      47.           }
      48.         else
      49.           {
      50.             Nil //全量函数的集合
      51.           }
      52.       }
      53.     )
      54.     val top10 = flatDatas.reduceByKey(
      55.       (t1 , t2) =>
      56.       {
      57.         (t1._1 + t2._1 ,t1._2+t2._2 ,t1._3+t2._3)
      58.       }
      59.     ).sortBy(_._2,false).take(10)
      60.  
      61.     //TODO 将统计采集之后的结果打印到控制台上
      62.     top10.foreach(println)
      63.  
      64.     sc.stop()
      65.   }
      66.  
      67. }

      1.5 实现方案四

      1.5.1 需求分析

      使用累加器的过程

      1.5.2 需求实现

    • 实现代码:比较难一些 
      1. package com.atguigu.bigdata.spark.req
      2.  
      3. import org.apache.spark.util.AccumulatorV2
      4. import org.apache.spark.{SparkConf, SparkContext}
      5.  
      6. import scala.collection.mutable
      7.  
      8. object Spark01_Req_HotCategoryTop10_3 {
      9.  
      10.     def main(args: Array[String]): Unit = {
      11.  
      12.         val conf = new SparkConf().setMaster("local[*]").setAppName("HotCategoryTop10")
      13.         val sc = new SparkContext(conf)
      14.  
      15.         val fileDatas = sc.textFile("data/user_visit_action.txt")
      16.  
      17.         // 创建累加器对象
      18.         val acc = new HotCategoryAccumulator()
      19.         // 注册累加器
      20.         sc.register(acc, "HotCategory")
      21.  
      22.         fileDatas.foreach(
      23.             data => {
      24.                 val datas = data.split("_")
      25.                 if ( datas(6) != "-1" ) {
      26.                     // 点击的场合
      27.                     acc.add( (datas(6), "click") )
      28.                 } else if ( datas(8) != "null" ) {
      29.                     // 下单的场合
      30.                     val id = datas(8)
      31.                     val ids = id.split(",")
      32.                     ids.foreach(
      33.                         id => {
      34.                             acc.add( (id, "order") )
      35.                         }
      36.                     )
      37.                 } else if ( datas(10) != "null" ) {
      38.                     // 支付的场合
      39.                     val id = datas(10)
      40.                     val ids = id.split(",")
      41.                     ids.foreach(
      42.                         id => {
      43.                             acc.add( (id, "pay") )
      44.                         }
      45.                     )
      46.                 }
      47.             }
      48.         )
      49.  
      50.         // TODO 获取累加器的结果
      51.         val resultMap: mutable.Map[String, HotCategoryCount] = acc.value
      52.         val top10 = resultMap.map(_._2).toList.sortWith(
      53.             (left, right) => {
      54.                 if ( left.clickCnt > right.clickCnt ) {
      55.                     true
      56.                 } else if ( left.clickCnt == right.clickCnt ) {
      57.                     if ( left.orderCnt > right.orderCnt ) {
      58.                         true
      59.                     } else if ( left.orderCnt == right.orderCnt ) {
      60.                         left.payCnt > right.payCnt
      61.                     } else {
      62.                         false
      63.                     }
      64.                 } else {
      65.                     false
      66.                 }
      67.             }
      68.         ).take(10)
      69.  
      70.         top10.foreach(println)
      71.  
      72.         sc.stop()
      73.  
      74.     }
      75.     case class HotCategoryCount( cid:String, var clickCnt : Int, var orderCnt : Int, var payCnt : Int )
      76.     // TODO 自定义热门点击累加器
      77.     //  1. 继承AccumulatorV2
      78.     //  2. 定义泛型
      79.     //     IN : (品类ID，行为类型)
      80.     //     OUT : Map[品类ID, HotCategoryCount]
      81.     //  3. 重写方法 （3 + 3）
      82.     class HotCategoryAccumulator extends AccumulatorV2[(String, String), mutable.Map[String, HotCategoryCount]]{
      83.  
      84.         private val map = mutable.Map[String, HotCategoryCount]()
      85.  
      86.         override def isZero: Boolean = {
      87.             map.isEmpty
      88.         }
      89.  
      90.         override def copy(): AccumulatorV2[(String, String), mutable.Map[String, HotCategoryCount]] = {
      91.             new HotCategoryAccumulator()
      92.         }
      93.  
      94.         override def reset(): Unit = {
      95.             map.clear()
      96.         }
      97.  
      98.         override def add(v: (String, String)): Unit = {
      99.             val (cid, actionType) = v
      100.             val hcc: HotCategoryCount = map.getOrElse(cid, HotCategoryCount(cid, 0, 0, 0))
      101.             actionType match {
      102.                 case "click" => hcc.clickCnt += 1
      103.                 case "order" => hcc.orderCnt += 1
      104.                 case "pay" => hcc.payCnt += 1
      105.             }
      106.             map.update(cid, hcc)
      107.         }
      108.  
      109.         override def merge(other: AccumulatorV2[(String, String), mutable.Map[String, HotCategoryCount]]): Unit = {
      110.             other.value.foreach {
      111.                 case ( cid, otherHCC ) => {
      112.                     val thisHCC: HotCategoryCount = map.getOrElse(cid, HotCategoryCount(cid, 0, 0, 0))
      113.                     thisHCC.clickCnt += otherHCC.clickCnt
      114.                     thisHCC.orderCnt += otherHCC.orderCnt
      115.                     thisHCC.payCnt += otherHCC.payCnt
      116.                     map.update(cid, thisHCC)
      117.                 }
      118.             }
      119.         }
      120.  
      121.         override def value: mutable.Map[String, HotCategoryCount] = {
      122.             map
      123.         }
      124.     }
      125. }

    需求2：Top10热门品类中每个品类的Top10活跃Session统计

    2.1 需求说明

    在需求一的基础上，增加每个品类用户session的点击统计

    2.2需求分析

    JavaEE        Web:浏览器(zhangsan) => 服务器(lisi) [lisi懵了]       Session:会话(通信状态)

    2.3 功能实现(这里代码之中的架构需要进行注意)

    HotCataegoryTop10的代码 

    • Common之中的代码,用来呈递上面架构之中的架构部分,对里面的代码进行相应的封装:

    ①对于Application部分的封装

    1. package com.atguigu.bigdata.spark.summer.common
    2.  
    3. import com.atguigu.bigdata.spark.summer.util.EnvCache
    4. import org.apache.spark.{SparkConf, SparkContext}
    5.  
    6.  
    7. trait TApplication {
    8.  
    9.     //下面第一个括号之中代表的是参数，第二个代表的是进行执行的操作
    10.     def execute(master:String = "local[*]", appName:String)( op: =>Unit ): Unit = {
    11.         
    12.         val conf: SparkConf = new SparkConf().setMaster(master).setAppName(appName)
    13.         val sc = new SparkContext(conf)
    14.         EnvCache.put(sc) //用来存放sc这个值,后面可以拿来用
    15.         try {
    16.             op
    17.         } catch {
    18.             case e: Exception => e.printStackTrace()
    19.         }
    20.  
    21.         sc.stop()
    22.         EnvCache.clear()
    23.     }
    24. }

    ②对于Controller部分的封装

    1. package com.atguigu.bigdata.spark.summer.common
    2.  
    3. trait TController {
    4.  
    5.     def dispatch(): Unit
    6. }

    ③对于Dao部分的封装

    1. package com.atguigu.bigdata.spark.summer.common
    2.  
    3.  
    4. import com.atguigu.bigdata.spark.summer.util.EnvCache
    5. import org.apache.spark.SparkContext
    6.  
    7. import scala.io.{BufferedSource, Source}
    8.  
    9. trait TDao {
    10.     def readFile(  path : String ) = {
    11.         // e:/data/word.txt
    12.         val source: BufferedSource = Source.fromFile(EnvCache.get() + path)
    13.         val lines = source.getLines().toList
    14.         source.close()
    15.         lines
    16.     }
    17.     def readFileBySpark(  path : String ) = {
    18.         //对于前面使用的sc进行一个读取的过程
    19.         EnvCache.get().asInstanceOf[SparkContext].textFile(path)
    20.     }
    21. }

    ④Service部分的封装

    1. package com.atguigu.bigdata.spark.summer.common
    2.  
    3. trait TService {
    4.     def analysis() : Any = {
    5.  
    6.     }
    7.     def analysis( data : Any ) : Any = {
    8.  
    9.     }
    10. }
    • Application(应用的起点)部分的代码:
    1. package com.atguigu.bigdata.spark.summer.application
    2.  
    3. import com.atguigu.bigdata.spark.summer.common.TApplication
    4. import com.atguigu.bigdata.spark.summer.controller.{HotCategoryTop10Controller, WordCountController}
    5.  
    6. //这里进行继承App便是可以之间执行的主函数，可以不用写出main函数
    7. object HotCategoryTop10Application extends TApplication with App{
    8.  
    9. //这里的execute本来是两个参数，但是其中一个是已经定义好的默认参数
    10.     execute(appName = "HotCategoryTop10"){
    11.       //执行controller
    12.         val controller = new HotCategoryTop10Controller
    13.         controller.dispatch()
    14.     }
    15. }
    • Controller(进行调度)部分的代码:
    1. package com.atguigu.bigdata.spark.summer.controller
    2.  
    3. import com.atguigu.bigdata.spark.summer.common.TController
    4. import com.atguigu.bigdata.spark.summer.service.HotCategoryTop10Service
    5.  
    6. class HotCategoryTop10Controller extends TController {
    7.  
    8.     private val hotCategoryTop10Service = new HotCategoryTop10Service
    9.  
    10.     override def dispatch(): Unit = {
    11.         val result: Array[(String, (Int, Int, Int))] = hotCategoryTop10Service.analysis()
    12.         result.foreach(println)
    13.     }
    14. }
    • Dao(跟数据打交道)部分的代码
    1. package com.atguigu.bigdata.spark.summer.dao
    2.  
    3. import com.atguigu.bigdata.spark.summer.common.TDao
    4.  
    5. class HotCategoryTop10Dao  extends TDao  {
    6.  
    7. }
    • Service(执行逻辑)部分的代码
    1. package com.atguigu.bigdata.spark.summer.service
    2.  
    3. import com.atguigu.bigdata.spark.summer.common.TService
    4. import com.atguigu.bigdata.spark.summer.dao.HotCategoryTop10Dao
    5.  
    6. class HotCategoryTop10Service  extends TService {
    7.  
    8.     private val hotCategoryTop10Dao = new HotCategoryTop10Dao
    9.  
    10.     override def analysis() = {
    11.         val fileDatas = hotCategoryTop10Dao.readFileBySpark("data/user_visit_action.txt")
    12.         val flatDatas = fileDatas.flatMap(
    13.             data => {
    14.                 var datas = data.split("_")
    15.                 if ( datas(6) != "-1" ) {
    16.                     // 点击数据的场合
    17.                     List((datas(6), (1, 0, 0)))
    18.                 } else if ( datas(8) != "null" ) {
    19.                     // 下单数据的场合
    20.                     val id = datas(8)
    21.                     val ids = id.split(",")
    22.                     ids.map(
    23.                         id => {
    24.                             (id, (0, 1, 0))
    25.                         }
    26.                     )
    27.                 } else if ( datas(10) != "null" ) {
    28.                     // 支付数据的场合
    29.                     val id = datas(10)
    30.                     val ids = id.split(",")
    31.                     ids.map(
    32.                         id => {
    33.                             (id, (0, 0, 1))
    34.                         }
    35.                     )
    36.                 } else {
    37.                     Nil
    38.                 }
    39.             }
    40.         )
    41.  
    42.         val top10 = flatDatas.reduceByKey(
    43.             (t1, t2) => {
    44.                 ( t1._1 + t2._1, t1._2 + t2._2, t1._3 + t2._3 )
    45.             }
    46.         ).sortBy(_._2, false).take(10)
    47.         top10
    48.     }
    49. }
    • Util(主线程之中用来存放rootPath连接)的代码:
    1. package com.atguigu.bigdata.spark.summer.util
    2.  
    3. object EnvCache {
    4.  
    5.     private val envCache : ThreadLocal[Object] = new ThreadLocal[Object]
    6.  
    7.     def put( data : Object ): Unit = {
    8.         envCache.set(data)
    9.     }
    10.  
    11.     def get() = {
    12.         envCache.get()
    13.     }
    14.  
    15.     def clear(): Unit = {
    16.         envCache.remove()
    17.     }
    18. }

    HotCategoryTop10SessionController

    这部分的代码是在上面的原有基础上进行一个增加的过程.

    •  Application(应用的起点)部分的代码:
    1. package com.atguigu.bigdata.spark.summer.application
    2.  
    3. import com.atguigu.bigdata.spark.summer.common.TApplication
    4. import com.atguigu.bigdata.spark.summer.controller.{HotCategoryTop10Controller, HotCategoryTop10SessionController}
    5.  
    6.  
    7. object HotCategoryTop10SessionApplication extends TApplication with App{
    8.  
    9.     execute(appName = "HotCategoryTop10Session"){
    10.         val controller = new HotCategoryTop10SessionController
    11.         controller.dispatch()
    12.     }
    13. }
    • Controller(进行调度)部分的代码:
    1. package com.atguigu.bigdata.spark.summer.controller
    2.  
    3. import com.atguigu.bigdata.spark.summer.common.TController
    4. import com.atguigu.bigdata.spark.summer.service.{HotCategoryTop10Service, HotCategoryTop10SessionService}
    5.  
    6. class HotCategoryTop10SessionController extends TController {
    7.  
    8.     //由于这里是需要用到上面的逻辑，因此这里是存在两个代码
    9.     private val hotCategoryTop10Service = new HotCategoryTop10Service
    10.     private val hotCategoryTop10SessionService = new HotCategoryTop10SessionService
    11.  
    12.     override def dispatch(): Unit = {
    13.         val top10: Array[(String, (Int, Int, Int))] = hotCategoryTop10Service.analysis()
    14.         val result = hotCategoryTop10SessionService.analysis(top10.map(_._1))//只是存取Id
    15.  
    16.         result.foreach(println)
    17.     }
    18. }
    • Dao(跟数据打交道)部分的代码
    1. package com.atguigu.bigdata.spark.summer.dao
    2.  
    3. import com.atguigu.bigdata.spark.summer.common.TDao
    4.  
    5. class HotCategoryTop10SessionDao extends TDao  {
    6.  
    7. }
    • Service(执行逻辑)部分的代码
    1. package com.atguigu.bigdata.spark.summer.service
    2.  
    3. import com.atguigu.bigdata.spark.summer.bean.UserVisitAction
    4. import com.atguigu.bigdata.spark.summer.common.TService
    5. import com.atguigu.bigdata.spark.summer.dao.{HotCategoryTop10Dao, HotCategoryTop10SessionDao}
    6. import org.apache.spark.rdd.RDD
    7.  
    8. class HotCategoryTop10SessionService extends TService {
    9.  
    10.     private val hotCategoryTop10SessionDao = new HotCategoryTop10SessionDao
    11.  
    12.     override def analysis( data : Any ) = {
    13.         val topIds: Array[String] = data.asInstanceOf[Array[String]]
    14.  
    15.         val fileDatas = hotCategoryTop10SessionDao.readFileBySpark("data/user_visit_action.txt")
    16.  
    17.         val actionDatas = fileDatas.map(
    18.             data => {
    19.                 val datas = data.split("_")
    20.                 //将字符串转换为long类型
    21.                 UserVisitAction(
    22.                     datas(0),
    23.                     datas(1).toLong,
    24.                     datas(2),
    25.                     datas(3).toLong,
    26.                     datas(4),
    27.                     datas(5),
    28.                     datas(6).toLong,
    29.                     datas(7).toLong,
    30.                     datas(8),
    31.                     datas(9),
    32.                     datas(10),
    33.                     datas(11),
    34.                     datas(12).toLong
    35.                 )
    36.             }
    37.         )
    38.  
    39.         val clickDatas = actionDatas.filter {
    40.             data => {
    41.                 if ( data.click_category_id != -1 ) //点击不是-1
    42.                 {
    43.                     topIds.contains(data.click_category_id.toString)
    44.                 } else {
    45.                     false
    46.                 }
    47.             }
    48.         }
    49.  
    50.         val reduceDatas = clickDatas.map(
    51.             data => {
    52.                 (( data.click_category_id, data.session_id ), 1)
    53.             }
    54.         ).reduceByKey(_+_)
    55.  
    56.         val groupDatas: RDD[(Long, Iterable[(String, Int)])] = reduceDatas.map {
    57.             case ((cid, sid), cnt) => {
    58.                 (cid, (sid, cnt))
    59.             }
    60.         }.groupByKey()
    61.  
    62.         groupDatas.mapValues(
    63.             iter => {
    64.                 iter.toList.sortBy(_._2)(Ordering.Int.reverse).take(10)
    65.             }
    66.         ).collect()
    67.  
    68.     }
    69. }
    • Bean之中增加UserVisitAcition代码如下所示:
    1. package com.atguigu.bigdata.spark.summer.bean
    2.  
    3. //用户访问动作表
    4. case class UserVisitAction(
    5.   date: String,//用户点击行为的日期
    6.   user_id: Long,//用户的ID
    7.   session_id: String,//Session的ID
    8.   page_id: Long,//某个页面的ID
    9.   action_time: String,//动作的时间点
    10.   search_keyword: String,//用户搜索的关键词
    11.   click_category_id: Long,//某一个商品品类的ID
    12.   click_product_id: Long,//某一个商品的ID
    13.   order_category_ids: String,//一次订单中所有品类的ID集合
    14.   order_product_ids: String,//一次订单中所有商品的ID集合
    15.   pay_category_ids: String,//一次支付中所有品类的ID集合
    16.   pay_product_ids: String,//一次支付中所有商品的ID集合
    17.   city_id: Long//城市 id
    18. )

    需求3：页面单跳转换率统计

    3.1 需求说明

    1）页面单跳转化率

    计算页面单跳转化率，什么是页面单跳转换率，比如一个用户在一次 Session 过程中访问的页面路径 3,5,7,9,10,21，那么页面 3 跳到页面 5 叫一次单跳，7-9 也叫一次单跳，那么单跳转化率就是要统计页面点击的概率。

    比如：计算 3-5 的单跳转化率，先获取符合条件的 Session 对于页面 3 的访问次数（PV）为 A，然后获取符合条件的 Session 中访问了页面 3 又紧接着访问了页面 5 的次数为 B，那么 B/A 就是 3-5 的页面单跳转化率。

    为什么IE被淘汰了?因为IE不支持新的技术,而且不能够装入插件.不要说什么产品经理没用,你是可以替换的,但是产品经理是不可以替换的.

    2）统计页面单跳转化率意义

    产品经理和运营总监，可以根据这个指标，去尝试分析，整个网站，产品，各个页面的表现怎么样，是不是需要去优化产品的布局；吸引用户最终可以进入最后的支付页面。

    数据分析师，可以此数据做更深一步的计算和分析。

    企业管理层，可以看到整个公司的网站，各个页面的之间的跳转的表现如何，可以适当调整公司的经营战略或策略。

    信息产生过剩:产生的大量的垃圾信息,在垃圾信息之中,找出有用的信息.

    3.2 需求分析

     第一列是进行一个采集的过程,数据产生乱序的原因,如下流程图所示:

     上述过程是需要进行计算一个分子与分母的过程,分子(首页 - 详情)  分母(首页)

    1. package com.atguigu.bigdata.spark.req
    2.  
    3. import com.atguigu.bigdata.spark.summer.bean.UserVisitAction
    4.  
    5. object Spark03_Req_PageFlow  {
    6.  
    7.     def main(args: Array[String]): Unit = {
    8.  
    9.         val conf = new SparkConf().setMaster("local[*]").setAppName("Pageflow")
    10.         val sc = new SparkContext(conf)
    11.  
    12.         val fileDatas = sc.textFile("data/user_visit_action.txt")
    13.  
    14.         //包装成为样例类
    15.         val actionDatas = fileDatas.map(
    16.             data => {
    17.                 val datas = data.split("_")
    18.                 UserVisitAction(
    19.                     datas(0),
    20.                     datas(1).toLong,
    21.                     datas(2),
    22.                     datas(3).toLong,
    23.                     datas(4),
    24.                     datas(5),
    25.                     datas(6).toLong,
    26.                     datas(7).toLong,
    27.                     datas(8),
    28.                     datas(9),
    29.                     datas(10),
    30.                     datas(11),
    31.                     datas(12).toLong
    32.                 )
    33.             }
    34.         )
    35.         actionDatas.cache()//做一个缓存过程
    36.  
    37.         // 【1，2，3，4，5，6，7】
    38.         val okIds = List(1,2,3,4,5,6,7)
    39.         // 【（1，2），（2，3）】
    40.         val okFlowIds = okIds.zip(okIds.tail)
    41.  
    42.         // TODO 分母的计算 Long是页面id，int是统计结果
    43.         val result: Map[Long, Int] = actionDatas.filter(
    44.             action => {
    45.                 okIds.init.contains(action.page_id.toInt)
    46.             }
    47.         ).map(
    48.             action => {
    49.                 (action.page_id, 1)
    50.             }
    51.         ).reduceByKey(_ + _).collect().toMap
    52.  
    53.         // TODO 分子的计算
    54.         // 1. 将数据按照session进行分组
    55.         val groupRDD: RDD[(String, Iterable[UserVisitAction])] = actionDatas.groupBy(_.session_id)
    56.  
    57.         // 将分组后的数据进行组内排序
    58.         val mapRDD = groupRDD.mapValues(
    59.             iter => {
    60.                 //toList含义就是将上面的东西进行转换，进行可以迭代操作
    61.                 val actions: List[UserVisitAction] = iter.toList.sortBy(_.action_time)
    62.                 //【1,2,3,4,5,6,7】
    63.                 //【2,3,4,5,6,7】
    64.                 // 滑窗
    65.                 //【1-2，2-3，3-4，4-5，5-6，6-7】
    66.                 val ids: List[Int] = actions.map(_.page_id.toInt)
    67.  
    68.                 //                val iterator: Iterator[List[Long]] = ids.sliding(2)
    69.                 //                while ( iterator.hasNext ) {
    70.                 //                    val longs: List[Long] = iterator.next()
    71.                 //                    (longs.head, longs.last)
    72.                 //                }
    73.                 
    74.                 val flowIds: List[(Int, Int)] = ids.zip(ids.tail)
    75.  
    76.                 flowIds.filter(
    77.                     ids => {
    78.                         okFlowIds.contains(ids)
    79.                     }
    80.                 )
    81.             }
    82.         )
    83.         val mapRDD2 = mapRDD.map(_._2)
    84.         val flatRDD = mapRDD2.flatMap(list => list)
    85.  
    86.         // 分子计算完毕
    87.         val reduceRDD = flatRDD.map((_, 1)).reduceByKey(_ + _)
    88.         // TODO 单挑转换率的统计
    89.         reduceRDD.foreach {
    90.             case ( (id1, id2), cnt ) => {
    91.                 println(s"页面【${id1}-${id2}】单挑转换率为 ：" + ( cnt.toDouble / result.getOrElse(id1, 1) ))
    92.             }
    93.         }
    94.  
    95.  
    96.         sc.stop()
    97.  
    98.     }
    99. }
  • 相关阅读:
    Docker - 容器的网络模式
    MySQL 数据库 查询定义参数【模糊查询】
    终于还是熬不住了，转行了，分享一波刚学到的知识吧,字符串的自带函数.py
    （免费领源码）java#SSM#Mysql学院教室管理系统81671-计算机毕业设计项目选题推荐
    vue -权限管理-指令-v-permission
    【MySQL入门实战5】-Linux PRM 包安装MySQL
    Java#14(StringJoiner)
    Vue 全局状态管理工具 pinia
    《Linux运维总结：内网服务器通过代理访问外网服务器（方法一）》
    python迭代器
  • 原文地址：https://blog.csdn.net/m0_47489229/article/details/125922298