Spark Core案例实操（四）

Leefs 2021-10-31 PM 2794℃ 0条

### 一、需求

> 页面单跳转换率统计

#### 需求说明

+ 计算页面单跳转化率，什么是页面单跳转换率，比如一个用户在一次 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 的页面单跳转化率。

![14.Spark Core案例实操（四）01.jpg](https://lilinchao.com/usr/uploads/2021/10/2222701000.jpg)

+ 在该模块中，需要根据查询对象中设置的 Session 过滤条件，先将对应得 Session 过滤出来，然后根据查询对象中设置的页面路径，计算页面单跳转化率，比如查询的页面路径为：3、5、7、8，那么就要计算 3-5、5-7、7-8 的页面单跳转化率。需要注意的一点是，**页面的访问时有先后的，要做好排序**。

### 二、实现

#### 2.1 思路分析

```
1.读取原始数据
2.将原始数据映射为样例类
3.将原始数据根据session进行分组
4.将分组后的数据根据时间进行排序（升序）
5.将排序后的数据进行结构的转换（pageId,1）
6.计算分母-将相同的页面id进行聚合统计(pageId,sum)
7.计算分子-将页面id进行拉链，形成连续的拉链效果，转换结构(pageId-pageId2,1)
8.将转换结构后的数据进行聚合统计(pageId-pageId2,sum)
9.计算页面单跳转换率
```

![14.Spark Core案例实操（四）02.png](https://lilinchao.com/usr/uploads/2021/10/1234232444.png)

#### 2.2 代码实现

```scala
package com.llc.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
  * @author lilinchao
  * @date 2021/10/31
  * @description 1.0 
  **/
object PageflowAnalysis {

def main(args: Array[String]): Unit = {

val sparConf = new SparkConf().setMaster("local[*]").setAppName("HotCategoryTop10Analysis")
    val sc = new SparkContext(sparConf)

// 1. 读取原始日志数据
    val actionRDD = sc.textFile("datas/user_visit_action.txt")

val actionDataRDD = actionRDD.map(
      action => {
        val datas = action.split("_")
        UserVisitAction(
          datas(0),
          datas(1).toLong,
          datas(2),
          datas(3).toLong,
          datas(4),
          datas(5),
          datas(6).toLong,
          datas(7).toLong,
          datas(8),
          datas(9),
          datas(10),
          datas(11),
          datas(12).toLong
        )
      }
    )
    actionDataRDD.cache()

// TODO 对指定的页面连接跳转进行统计
    // 1-2,2-3,3-4,4-5,5-6,6-7
    val ids = List[Long](1,2,3,4,5,6,7)
    // tail:获取除了第一个元素之外的所有元素
    // zip:拉链操作，用于关联两个集合。如果其中一个参数元素比较长，那么多余的参数会被删掉。
    val okflowIds: List[(Long, Long)] = ids.zip(ids.tail)

//    okflowIds.foreach(println)

//筛选过滤，对指定页面，根据页面ID进行聚合操作
    val pageidToCountMap: Map[Long, Long] = actionDataRDD.filter(
      action => {
        ids.init.contains(action.page_id)
      }
    ).map(
      action => {
        (action.page_id,1L)
      }
    ).reduceByKey(_ + _).collect().toMap

//    pageidToCountMap.foreach(println)
    // TODO 计算分子

//根据session进行分组
    val sessionRDD: RDD[(String, Iterable[UserVisitAction])] = actionDataRDD.groupBy(_.session_id)
    // 分组后，根据访问时间进行排序（排序）
    val mvRDD: RDD[(String, List[((Long, Long), Int)])] = sessionRDD.mapValues(
      iter => {
        val sortList: List[UserVisitAction] =  iter.toList.sortBy(_.action_time)

// 【1，2，3，4】
        // 【1，2】，【2，3】，【3，4】
        // 【1-2，2-3，3-4】
        // Sliding : 滑窗
        // 【1，2，3，4】
        // 【2，3，4】
        // zip : 拉链
        val flowIds: List[Long] = sortList.map(_.page_id)
        val pageflowIds: List[(Long, Long)] = flowIds.zip(flowIds.tail)

// 将不合法的页面跳转进行过滤
        pageflowIds.filter(
          t => {
            okflowIds.contains(t)
          }
        ).map(
          t => {
            (t,1)
          }
        )
      }
    )

// ((1,2),1)
    val flatRDD: RDD[((Long, Long), Int)] = mvRDD.map(_._2).flatMap(list=>list)
//    flatRDD.foreach(println)
    // ((1,2),1) => ((1,2),sum)
    val dataRDD = flatRDD.reduceByKey(_+_)

// TODO 计算单跳转换率
    //分子除以分母
    dataRDD.foreach{
      case ((pageid1,pageid2),sum) => {
        val lon:Long = pageidToCountMap.getOrElse(pageid1,0L)

println(s"页面${pageid1}跳转到页面${pageid2}单跳转换率为:" + ( sum.toDouble/lon ))
      }
    }

sc.stop()
  }

//用户访问动作表
  case class UserVisitAction(
        date: String,//用户点击行为的日期
        user_id: Long,//用户的ID
        session_id: String,//Session的ID
        page_id: Long,//某个页面的ID
        action_time: String,//动作的时间点
        search_keyword: String,//用户搜索的关键词
        click_category_id: Long,//某一个商品品类的ID
        click_product_id: Long,//某一个商品的ID
        order_category_ids: String,//一次订单中所有品类的ID集合
        order_product_ids: String,//一次订单中所有商品的ID集合
        pay_category_ids: String,//一次支付中所有品类的ID集合
        pay_product_ids: String,//一次支付中所有商品的ID集合
        city_id: Long
  )//城市 id
}
```

**运行结果**

```
页面2跳转到页面3单跳转换率为:0.019949423995504357
页面1跳转到页面2单跳转换率为:0.01510989010989011
页面5跳转到页面6单跳转换率为:0.014594442885209093
页面4跳转到页面5单跳转换率为:0.018323153803442533
页面3跳转到页面4单跳转换率为:0.016884531590413945
页面6跳转到页面7单跳转换率为:0.0192040077929307
```

### 结尾

因为本篇使用的示例数据`user_visit_action.txt`文件由于数据量较大将不在下方贴出。

直接在微信公众号【Java和大数据进阶】回复：**sparkdata**，即可获取。

*附参考文章链接：*

*https://www.cxymm.net/article/qq_38689352/117063116*

标签: Spark, Spark Core

非特殊说明，本博所有文章均为博主原创。

如若转载，请注明出处：https://lilinchao.com/archives/1600.html

上一篇 Spark Core案例实操（三）

下一篇 Spark Core案例实操（五）

Spark Core案例实操（四）

评论啦~

栏目分类

标签云

友情链接申请

Spark Core案例实操（四）

 评论啦~

 栏目分类

标签云

友情链接申请

评论啦~

栏目分类

标签云

友情链接申请