Spark詳解(05-1) - SparkCore實戰案例

Spark詳解(05-1) - SparkCore實戰案例

數據准備

1）數據格式

本項目的數據是采集電商網站的用戶行為數據，主要包含用戶的4種行為：搜索、點擊、下單和支付。

（1）數據采用_分割字段

（2）每一行表示用戶的一個行為，所以每一行只能是四種行為中的一種。

（3）如果點擊的品類id和產品id是-1表示這次不是點擊

（4）針對下單行為，一次可以下單多個產品，所以品類id和產品id都是多個，id之間使用逗號分割。

如果本次不是下單行為，則他們相關數據用null來表示。

（5）支付行為和下單行為格式類似

2）數據詳細字段說明

編號	字段名稱	字段類型	字段含義
1	date	String	用戶點擊行為的日期
2	user_id	Long	用戶的ID
3	session_id	String	Session的ID
4	page_id	Long	某個頁面的ID
5	action_time	String	動作的時間點
6	search_keyword	String	用戶搜索的關鍵詞
7	click_category_id	Long	點擊某一個商品品類的ID
8	click_product_id	Long	某一個商品的ID
9	order_category_ids	String	一次訂單中所有品類的ID集合
10	order_product_ids	String	一次訂單中所有商品的ID集合
11	pay_category_ids	String	一次支付中所有品類的ID集合
12	pay_product_ids	String	一次支付中所有商品的ID集合
13	city_id	Long	城市 id

需求1：Top10熱門品類

需求說明：品類是指產品的分類，大型電商網站品類分多級，本項目中品類只有一級，不同的公司可能對熱門的定義不一樣。按照每個品類的點擊、下單、支付的量來統計熱門品類。

鞋            點擊數 下單數 支付數

衣服        點擊數 下單數 支付數

電腦        點擊數 下單數 支付數

例如，綜合排名 = 點擊數*20% + 下單數*30% + 支付數*50%

本項目需求優化為：先按照點擊數排名，靠前的就排名高；如果點擊數相同，再比較下單數；下單數再相同，就比較支付數。

需求分析（方案一）分步計算

思路：分別統計每個品類點擊的次數，下單的次數和支付的次數。

（品類，點擊總數）（品類，下單總數）（品類，支付總數）

缺點：統計3次，需要啟動3個job，每個job都有對原始數據遍歷一次，效率低。

需求分析（方案二）常規算子

采用常規算子的方式實現。

需求分析（方案三）樣例類

采用樣例類的方式實現。

需求實現（方案三）

1）用來封裝用戶行為的樣例類

1. //用戶訪問動作表
2. case class UserVisitAction(date: String,//用戶點擊行為的日期
3.                            user_id: Long,//用戶的ID
4.                            session_id: String,//Session的ID
5.                            page_id: Long,//某個頁面的ID
6.                            action_time: String,//動作的時間點
7.                            search_keyword: String,//用戶搜索的關鍵詞
8.                            click_category_id: Long,//某一個商品品類的ID
9.                            click_product_id: Long,//某一個商品的ID
10.                            order_category_ids: String,//一次訂單中所有品類的ID集合
11.                            order_product_ids: String,//一次訂單中所有商品的ID集合
12.                            pay_category_ids: String,//一次支付中所有品類的ID集合
13.                            pay_product_ids: String,//一次支付中所有商品的ID集合
14.                            city_id: Long)//城市 id
15. // 輸出結果表
16. case class CategoryCountInfo(categoryId: String,//品類id
17.                              clickCount: Long,//點擊次數
18.                              orderCount: Long,//訂單次數
19.                              payCount: Long)//支付次數

注意：樣例類的屬性默認是val修飾，不能修改；需要修改屬性，需要采用var修飾。

1. case class CategoryCountInfo(var categoryId: String,//品類id
2.                              var clickCount: Long,//點擊次數
3.                              var orderCount: Long,//訂單次數
4.                              var payCount: Long)//支付次數

2）核心業務代碼實現

1. object require01_top10Category_method3 {
2.  
3.     def main(args: Array[String]): Unit = {
4.  
5.         //1.創建SparkConf並設置App名稱
6.         val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[1]")
7.         //2.創建SparkContext，該對象是提交Spark App的入口
8.         val sc: SparkContext = new SparkContext(conf)
9.  
10.         //3 需求一：top10熱門品類
11.         //3.1 獲取原始數據
12.         val lineRDD: RDD[String] = sc.textFile("input/user_visit_action.txt")
13.         //3.2 將原始數據進行轉換（分解）
14.         val actionRDD: RDD[UserVisitAction] = lineRDD.map(
15.             line => {
16.                 // 獲取一行數據
17.                 val datas: Array[String] = line.split("_")
18.                 // 將解析的數據封裝到 UserVisitAction
19.                 UserVisitAction(
20.                     datas(0),
21.                     datas(1).toLong,
22.                     datas(2),
23.                     datas(3).toLong,
24.                     datas(4),
25.                     datas(5),
26.                     datas(6).toLong,
27.                     datas(7).toLong,
28.                     datas(8),
29.                     datas(9),
30.                     datas(10),
31.                     datas(11),
32.                     datas(12).toLong
33.                 )
34.             }
35.         )
36.  
37.         //CategoryCountInfo(鞋,1,0,0)
38.         //CategoryCountInfo(鞋,0,1,0)
39.         //CategoryCountInfo(鞋,0,0,1)
40.         //=>希望變成：CategoryCountInfo(鞋,1,1,1)
41.         //3.3 將轉換結構后的數據進行分解成CategoryCountInfo
42.         val infoRDD: RDD[CategoryCountInfo] = actionRDD.flatMap {
43.             case act: UserVisitAction => {
44.                 if (act.click_category_id != -1) { // 點擊信息處理
45.                     List(CategoryCountInfo(act.click_category_id.toString, 1, 0, 0))
46.                 } else if (act.order_category_ids != "null") { // 訂單信息處理
47.                     val list: ListBuffer[CategoryCountInfo] = new ListBuffer[CategoryCountInfo]
48.                     val ids: Array[String] = act.order_category_ids.split(",")
49.                     for (id <- ids) {
50.                         list.append(CategoryCountInfo(id, 0, 1, 0))
51.                     }
52.                     list
53.                 } else if (act.pay_category_ids != "null") { // 支付信息處理
54.                     val list: ListBuffer[CategoryCountInfo] = new ListBuffer[CategoryCountInfo]
55.                     val ids: Array[String] = act.pay_category_ids.split(",")
56.                     for (id <- ids) {
57.                         list.append(CategoryCountInfo(id, 0, 0, 1))
58.                     }
59.                     list
60.                 } else {
61.                     Nil
62.                 }
63.             }
64.         }
65.  
66.         //3.4 將相同的品類分成一組
67.         val groupRDD: RDD[(String, Iterable[CategoryCountInfo])] = infoRDD.groupBy(info => info.categoryId)
68.         //3.5 將分組后的數據進行聚合處理: (品類id, (品類id, clickCount, OrderCount, PayCount))
69.         val mapRDD: RDD[CategoryCountInfo] = groupRDD.mapValues(
70.  
71.             datas => {
72.                 datas.reduce(
73.                     (info1, info2) => {
74.                         info1.orderCount = info1.orderCount + info2.orderCount
75.                         info1.clickCount = info1.clickCount + info2.clickCount
76.                         info1.payCount = info1.payCount + info2.payCount
77.                         info1
78.                     }
79.                 )
80.             }
81.         ).map(_._2)
82.  
83.         //3.6 將聚合后的數據排序，取前10名
84.         val sortRDD: RDD[CategoryCountInfo] = mapRDD.sortBy(info => (info.clickCount, info.orderCount, info.payCount), false)
85.         val takeRDD: Array[CategoryCountInfo] = sortRDD.take(10)
86.  
87.         //3.7 打印
88.         takeRDD.foreach(println)
89.         //4.關閉連接
90.         sc.stop()
91.     }
92. }

需求分析（方案四）樣例類+算子優化

針對方案三中的groupBy，沒有提前聚合的功能，替換成reduceByKey

需求實現（方案四）

1）樣例類代碼和方案三一樣。（詳見方案三）

2）核心代碼實現

1. object require01_top10Category_method4 {
2.  
3.     def main(args: Array[String]): Unit = {
4.  
5.         //1.創建SparkConf並設置App名稱
6.         val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")
7.  
8.         //2.創建SparkContext，該對象是提交Spark App的入口
9.         val sc: SparkContext = new SparkContext(conf)
10.  
11.  
12.         //3.1 獲取原始數據
13.         val dataRDD: RDD[String] = sc.textFile("input/user_visit_action.txt")
14.  
15.         //3.2 將原始數據進行轉換
16.         val actionRDD: RDD[UserVisitAction] = dataRDD.map {
17.             data => {
18.                 val datas: Array[String] = data.split("_")
19.  
20.                 UserVisitAction(
21.                     datas(0),
22.                     datas(1).toLong,
23.                     datas(2),
24.                     datas(3).toLong,
25.                     datas(4),
26.                     datas(5),
27.                     datas(6).toLong,
28.                     datas(7).toLong,
29.                     datas(8),
30.                     datas(9),
31.                     datas(10),
32.                     datas(11),
33.                     datas(12).toLong
34.                 )
35.             }
36.         }
37.         //3.3 將轉換結構后的數據進行分解成：(品類,CategoryCountInfo)
38.         val infoRDD: RDD[(String, CategoryCountInfo)] = actionRDD.flatMap {
39.             action => {
40.                 action match {
41.                     case act: UserVisitAction => {
42.                         if (act.click_category_id != -1) {
43.                             List((act.click_category_id.toString, CategoryCountInfo(act.click_category_id.toString, 1, 0, 0)))
44.                         } else if (act.order_category_ids != "null") {
45.                             val list: ListBuffer[(String, CategoryCountInfo)] = new ListBuffer[(String, CategoryCountInfo)]
46.  
47.                             val ids: Array[String] = act.order_category_ids.split(",")
48.  
49.                             for (id <- ids) {
50.                                 list.append((id, CategoryCountInfo(id, 0, 1, 0)))
51.                             }
52.                             list
53.                         } else if (act.pay_category_ids != "null") {
54.  
55.                             val list: ListBuffer[(String, CategoryCountInfo)] = new ListBuffer[(String, CategoryCountInfo)]
56.                             val ids: mutable.ArrayOps[String] = act.pay_category_ids.split(",")
57.  
58.                             for (id <- ids) {
59.                                 list.append((id, CategoryCountInfo(id, 0, 0, 1)))
60.                             }
61.  
62.                             list
63.                         } else {
64.                             Nil
65.                         }
66.                     }
67.                     case _ => Nil
68.                 }
69.             }
70.         }
71.  
72.         //3.4 按照品類id相同，進行兩兩聚合
73.         val mapRDD: RDD[CategoryCountInfo] = infoRDD.reduceByKey(
74.             (info1, info2) => {
75.                 info1.orderCount = info1.orderCount + info2.orderCount
76.                 info1.clickCount = info1.clickCount + info2.clickCount
77.                 info1.payCount = info1.payCount + info2.payCount
78.  
79.                 info1
80.             }
81.         ).map(_._2)
82.  
83.         //3.6 將聚合后的數據排序，取前10名
84.         val sortRDD: RDD[CategoryCountInfo] = mapRDD.sortBy(info =>(info.clickCount,info.orderCount,info.payCount),false)
85.         val takeRDD: Array[CategoryCountInfo] = sortRDD.take(10)
86.  
87.         //3.7 打印
88.         takeRDD.foreach(println)
89.  
90.         //4.關閉連接
91.         sc.stop()
92.     }
93. }

需求分析（方案五）累加器

需求實現（方案五）

1）累加器實現

1. //品類行為統計累加器
2. // ((鞋,click),1)
3. // ((鞋,order),1)
4. // ((鞋,pay),1)
5. // ((衣服,pay),1)
6. //3.3 繼承AccumulatorV2，聲明泛型
7. //3.4 重寫方法
8. class CategoryCountAccumulator extends AccumulatorV2[UserVisitAction, mutable.Map[(String, String), Long]] {
9.  
10.     var map = mutable.Map[(String, String), Long]()
11.  
12.     override def isZero: Boolean = map.isEmpty
13.  
14.     override def copy(): AccumulatorV2[UserVisitAction, mutable.Map[(String, String), Long]] = {
15.         new CategoryCountAccumulator()
16.     }
17.  
18.     override def reset(): Unit = map.clear()
19.  
20.     override def add(action: UserVisitAction): Unit = {
21.         // 鞋 click + 鞋 click ...+ 鞋 click => ((鞋,click),100)
22.         // 鞋 order + 鞋 order ...+ 鞋 order =>((鞋,order),50)
23.         if (action.click_category_id != -1) {
24.             val key = (action.click_category_id.toString, "click")
25.  
26.             map(key) = map.getOrElse(key, 0L) + 1L
27.         } else if (action.order_category_ids != "null") {
28.             val ids: Array[String] = action.order_category_ids.split(",")
29.  
30.             for (id <- ids) {
31.                 val key = (id, "order")
32.                 map(key) = map.getOrElse(key, 0L) + 1L
33.             }
34.         } else if (action.pay_category_ids != "null") {
35.             val ids: Array[String] = action.pay_category_ids.split(",")
36.  
37.             for (id <- ids) {
38.                 val key = (id, "pay")
39.                 map(key) = map.getOrElse(key, 0L) + 1L
40.             }
41.         }
42.     }
43.  
44.     override def merge(other: AccumulatorV2[UserVisitAction, mutable.Map[(String, String), Long]]): Unit = {
45.  
46.         other.value.foreach {
47.             case (category, count) => {
48.                 map(category) = map.getOrElse(category, 0L) + count
49.             }
50.         }
51.     }
52.  
53.     override def value: mutable.Map[(String, String), Long] = map
54. }

2）核心邏輯實現

1. object require01_top10Category_method5 {
2.  
3.     def main(args: Array[String]): Unit = {
4.  
5.         //1.創建SparkConf並設置App名稱
6.         val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")
7.  
8.         //2.創建SparkContext，該對象是提交Spark App的入口
9.         val sc: SparkContext = new SparkContext(conf)
10.  
11.         //3.1 獲取原始數據
12.         val lineRDD: RDD[String] = sc.textFile("input/user_visit_action.txt")
13.  
14.         //3.2 將原始數據進行轉換
15.         val actionRDD: RDD[UserVisitAction] = lineRDD.map {
16.             line => {
17.                 val datas: Array[String] = line.split("_")
18.  
19.                 UserVisitAction(
20.                     datas(0),
21.                     datas(1).toLong,
22.                     datas(2),
23.                     datas(3).toLong,
24.                     datas(4),
25.                     datas(5),
26.                     datas(6).toLong,
27.                     datas(7).toLong,
28.                     datas(8),
29.                     datas(9),
30.                     datas(10),
31.                     datas(11),
32.                     datas(12).toLong
33.                 )
34.             }
35.         }
36.  
37.         //3.5 創建累加器
38.         val acc: CategoryCountAccumulator = new CategoryCountAccumulator()
39.  
40.         //3.6 注冊累加器
41.         sc.register(acc, "CategoryCountAccumulator")
42.  
43.         //3.7 累加器添加數據
44.         actionRDD.foreach(action => acc.add(action))
45.  
46.         //3.8 獲取累加器的值
47.         // ((鞋,click),10)
48.         // ((鞋,order),5)
49.         // =>(鞋，(click,order,pay))=>CategoryCountInfo
50.         val accMap: mutable.Map[(String, String), Long] = acc.value
51.  
52.         // 3.9 將累加器的值進行結構的轉換
53.         val group: Map[String, mutable.Map[(String, String), Long]] = accMap.groupBy(_._1._1)
54.  
55.         val infoes: immutable.Iterable[CategoryCountInfo] = group.map {
56.             case (id, map) => {
57.                 val click = map.getOrElse((id, "click"), 0L)
58.                 val order = map.getOrElse((id, "order"), 0L)
59.                 val pay = map.getOrElse((id, "pay"), 0L)
60.  
61.                 CategoryCountInfo(id, click, order, pay)
62.             }
63.         }
64.  
65.         //3.10 將轉換后的數據進行排序（降序），取前10
66.         infoes.toList.sortWith(
67.             (left,right)=>{
68.                 if (left.clickCount > right.clickCount){
69.                     true
70.                 }else if(left.clickCount == right.clickCount){
71.                     if (left.orderCount > right.orderCount){
72.                         true
73.  
74.                     }else if(left.orderCount == right.orderCount){
75.                         left.payCount > right.payCount
76.                     }else {
77.                         false
78.                     }
79.                 }else{
80.                     false
81.                 }
82.             }
83.         ).take(10).foreach(println)
84.  
85.         //4.關閉連接
86.         sc.stop()
87.     }
88. }

需求2：Top10熱門品類中每個品類的Top10活躍Session統計

需求分析

需求實現

1）累加器實現

1. //品類行為統計累加器
2. // ((鞋,click),1)
3. // ((鞋,order),1)
4. // ((鞋,pay),1)
5. // ((衣服,pay),1)
6. //3.3 繼承AccumulatorV2，聲明泛型
7. //3.4 重寫方法
8. class CategoryCountAccumulator extends AccumulatorV2[UserVisitAction, mutable.Map[(String, String), Long]] {
9.  
10.     var map = mutable.Map[(String, String), Long]()
11.  
12.     override def isZero: Boolean = map.isEmpty
13.  
14.     override def copy(): AccumulatorV2[UserVisitAction, mutable.Map[(String, String), Long]] = {
15.         new CategoryCountAccumulator()
16.     }
17.  
18.     override def reset(): Unit = map.clear()
19.  
20.     override def add(action: UserVisitAction): Unit = {
21.         // 鞋 click + 鞋 click ...+ 鞋 click => ((鞋,click),100)
22.         // 鞋 order + 鞋 order ...+ 鞋 order =>((鞋,order),50)
23.         if (action.click_category_id != -1) {
24.             val key = (action.click_category_id.toString, "click")
25.  
26.             map(key) = map.getOrElse(key, 0L) + 1L
27.         } else if (action.order_category_ids != "null") {
28.             val ids: Array[String] = action.order_category_ids.split(",")
29.  
30.             for (id <- ids) {
31.                 val key = (id, "order")
32.                 map(key) = map.getOrElse(key, 0L) + 1L
33.             }
34.         } else if (action.pay_category_ids != "null") {
35.             val ids: Array[String] = action.pay_category_ids.split(",")
36.  
37.             for (id <- ids) {
38.                 val key = (id, "pay")
39.                 map(key) = map.getOrElse(key, 0L) + 1L
40.             }
41.         }
42.     }
43.  
44.     override def merge(other: AccumulatorV2[UserVisitAction, mutable.Map[(String, String), Long]]): Unit = {
45.  
46.         other.value.foreach {
47.             case (category, count) => {
48.                 map(category) = map.getOrElse(category, 0L) + count
49.             }
50.         }
51.     }
52.  
53.     override def value: mutable.Map[(String, String), Long] = map
54. }

2）核心邏輯實現

1. object require02_top10Category_sessionTop10 {
2.  
3.     def main(args: Array[String]): Unit = {
4.  
5.         //1.創建SparkConf並設置App名稱
6.         val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")
7.  
8.         //2.創建SparkContext，該對象是提交Spark App的入口
9.         val sc: SparkContext = new SparkContext(conf)
10.  
11.         //3.1 獲取原始數據
12.         val dataRDD: RDD[String] = sc.textFile("input/user_visit_action.txt")
13.  
14.         //3.2 將原始數據進行轉換
15.         val actionRDD: RDD[UserVisitAction] = dataRDD.map {
16.             data => {
17.                 val datas: Array[String] = data.split("_")
18.  
19.                 UserVisitAction(
20.                     datas(0),
21.                     datas(1).toLong,
22.                     datas(2),
23.                     datas(3).toLong,
24.                     datas(4),
25.                     datas(5),
26.                     datas(6).toLong,
27.                     datas(7).toLong,
28.                     datas(8),
29.                     datas(9),
30.                     datas(10),
31.                     datas(11),
32.                     datas(12).toLong
33.                 )
34.             }
35.         }
36.  
37.         //3.5 創建累加器
38.         val acc: CategoryCountAccumulator = new CategoryCountAccumulator()
39.  
40.         //3.6 注冊累加器
41.         sc.register(acc, "CategoryCountAccumulator")
42.  
43.         //3.7 累加器添加數據
44.         actionRDD.foreach(action => acc.add(action))
45.  
46.         //3.8 獲取累加器的值
47.         // ((鞋,click),10)
48.         // ((鞋,order),5)
49.         // =>(鞋，(click,order,pay))=>CategoryCountInfo
50.         val accMap: mutable.Map[(String, String), Long] = acc.value
51.  
52.         // 3.9 將累加器的值進行結構的轉換
53.         val group: Map[String, mutable.Map[(String, String), Long]] = accMap.groupBy(_._1._1)
54.  
55.         val infoes: immutable.Iterable[CategoryCountInfo] = group.map {
56.             case (id, map) => {
57.                 val click = map.getOrElse((id, "click"), 0L)
58.                 val order = map.getOrElse((id, "order"), 0L)
59.                 val pay = map.getOrElse((id, "pay"), 0L)
60.  
61.                 CategoryCountInfo(id, click, order, pay)
62.             }
63.         }
64.  
65.         //3.10 將轉換后的數據進行排序（降序），取前10
66.         val sort: List[CategoryCountInfo] = infoes.toList.sortWith(
67.             (left, right) => {
68.                 if (left.clickCount > right.clickCount) {
69.                     true
70.                 } else if (left.clickCount == right.clickCount) {
71.                     if (left.orderCount > right.orderCount) {
72.                         true
73.  
74.                     } else if (left.orderCount == right.orderCount) {
75.                         left.payCount > right.payCount
76.                     } else {
77.                         false
78.                     }
79.                 } else {
80.                     false
81.                 }
82.             }
83.         )
84.  
85.         val top10Info: List[CategoryCountInfo] = sort.take(10)
86.  
87.         //********************需求二********************************
88.         //4.1 獲取Top10熱門品類
89.         val ids: List[String] = top10Info.map(_.categoryId)
90.  
91.         //4.2 ids變成廣播變量
92.         val broadcastIds: Broadcast[List[String]] = sc.broadcast(ids)
93.  
94.         //4.3 將原始數據進行過濾（保留前10熱門品類的數據，保留點擊數據）
95.         val filterActionRDD: RDD[UserVisitAction] = actionRDD.filter(
96.             action => {
97.                 if (action.click_category_id != -1) {
98.                     broadcastIds.value.contains(action.click_category_id.toString)
99.                 } else {
100.                     false
101.                 }
102.             }
103.         )
104.  
105.         //4.4 對session點擊次數進行轉換：(categoryid-session, 1)
106.         val idAndSessionToOneRDD: RDD[(String, Int)] = filterActionRDD.map(
107.             action => (action.click_category_id + "--" + action.session_id, 1)
108.         )
109.  
110.         //4.5 對session點擊次數進行統計：(categoryid-session, sum)
111.         val idAndSessionToSumRDD: RDD[(String, Int)] = idAndSessionToOneRDD.reduceByKey(_+_)
112.  
113.         //4.6 將統計結果進行結構的轉換：(categoryid, (session,sum))
114.         val idToSessionAndSumRDD: RDD[(String, (String, Int))] = idAndSessionToSumRDD.map {
115.             case (key, sum) => {
116.                 val keys: Array[String] = key.split("--")
117.  
118.                 (keys(0), (keys(1), sum))
119.             }
120.         }
121.  
122.         //4.7 將轉換結構后的數據根據品類進行分組：(categoryid, Iterator[(session,sum)])
123.         val idToSessionAndSumGroupRDD: RDD[(String, Iterable[(String, Int)])] = idToSessionAndSumRDD.groupByKey()
124.  
125.         //4.8 將分組后的數據進行排序（降序），取前10名
126.         val resultRDD: RDD[(String, List[(String, Int)])] = idToSessionAndSumGroupRDD.mapValues {
127.             datas => {
128.                 datas.toList.sortWith(
129.                     (left, right) => {
130.                         left._2 > right._2
131.                     }
132.                 ).take(10)
133.             }
134.         }
135.         resultRDD.foreach(println)
136.  
137.         //5.關閉連接
138.         sc.stop()
139.     }
140. }

需求3：頁面單跳轉化率統計

需求分析

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 的頁面單跳轉化率。

2）統計頁面單跳轉化率意義

產品經理和運營總監，可以根據這個指標，去嘗試分析，整個網站，產品，各個頁面的表現怎么樣，是不是需要去優化產品的布局；吸引用戶最終可以進入最后的支付頁面。

數據分析師，可以此數據做更深一步的計算和分析。

企業管理層，可以看到整個公司的網站，各個頁面的之間的跳轉的表現如何，可以適當調整公司的經營戰略或策略。

3）需求詳細描述

在該模塊中，需要根據查詢對象中設置的Session過濾條件，先將對應得Session過濾出來，然后根據查詢對象中設置的頁面路徑，計算頁面單跳轉化率，比如查詢的頁面路徑為：3、5、7、8，那么就要計算3-5、5-7、7-8的頁面單跳轉化率。

需要注意的一點是，頁面的訪問是有先后的，要做好排序。

1、2、3、4、5、6、7

1-2/ 1 2-3/2 3-4/3 4-5/4 5-6/5 6-7/6

4）需求分析

用戶行為數據采集模擬

需求實現

1）代碼實現

1. object require03_PageFlow {
2.  
3.     def main(args: Array[String]): Unit = {
4.  
5.         //1.創建SparkConf並設置App名稱
6.         val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")
7.  
8.         //2.創建SparkContext，該對象是提交Spark App的入口
9.         val sc: SparkContext = new SparkContext(conf)
10.  
11.         //3.1 獲取原始數據
12.         val lineRDD: RDD[String] = sc.textFile("input/user_visit_action.txt")
13.  
14.         //3.2 將原始數據進行轉換
15.         val actionRDD: RDD[UserVisitAction] = lineRDD.map {
16.             line => {
17.                 val datas: Array[String] = line.split("_")
18.  
19.                 UserVisitAction(
20.                     datas(0),
21.                     datas(1).toLong,
22.                     datas(2),
23.                     datas(3).toLong,
24.                     datas(4),
25.                     datas(5),
26.                     datas(6).toLong,
27.                     datas(7).toLong,
28.                     datas(8),
29.                     datas(9),
30.                     datas(10),
31.                     datas(11),
32.                     datas(12).toLong
33.                 )
34.             }
35.         }
36.  
37.         //3.3 定義要統計的頁面（只統計集合中規定的頁面跳轉率）
38.         val ids = List(1, 2, 3, 4, 5, 6, 7)
39.         // 准備過濾數據
40.         val idZipList: List[String] = ids.zip(ids.tail).map {
41.             case (pageId1, pageId2) => {
42.                 pageId1 + "-" + pageId2
43.             }
44.         }
45.  
46.         //4 計算分母
47.         val idsMap: Map[Long, Long] = actionRDD
48.             // 過濾出要統計的page_id(由於最后一個頁面總次數，不參與運算，所以也過濾了)
49.             .filter(action => ids.init.contains(action.page_id))
50.             // 結構變換
51.             .map(action => (action.page_id, 1L))
52.             // 統計每個頁面的總次數
53.             .reduceByKey(_ + _).collect().toMap
54.  
55.         //5 計算分子
56.         //5.1 將原始數據根據Session進行分組：(session,Iterable[UserVisitAction])
57.         val sessionGroupRDD: RDD[(String, Iterable[UserVisitAction])] = actionRDD.groupBy(_.session_id)
58.  
59.         //5.2 將分組后的數據根據時間進行排序（升序）:List((pageId1-pageId2))
60.         val pageFlowRDD: RDD[List[String]] = sessionGroupRDD.mapValues(
61.             datas => {
62.                 //5.2.1 對分組后的數據進行排序
63.                 val actions: List[UserVisitAction] = datas.toList.sortWith(
64.                     (left, right) => {
65.                         left.action_time < right.action_time
66.                     }
67.                 )
68.  
69.                 //5.2.2 獲取PageId
70.                 val pageidList: List[Long] = actions.map(_.page_id)
71.  
72.                 //5.2.3 形成單跳元組(pageId1, pageId2)  (1,2,3,4…).zip((2,3,4…))
73.                 val pageToPageList: List[(Long, Long)] = pageidList.zip(pageidList.tail)
74.  
75.                 //5.2.4 變換結構
76.                 //=>List((pageId1-pageId2),(pageId2-pageId3),(pageId3-pageId4),(pageId4-pageId5),(pageId5-pageId6),(pageId6-pageId7))
77.                 val pageJumpCounts: List[String] = pageToPageList.map {
78.                     case (pageId1, pageId2) => {
79.                         pageId1 + "-" + pageId2
80.                     }
81.                 }
82.  
83.                 //5.2.5 再次進行過濾，減輕計算負擔
84.                 // 1-2 2-3 3-4 4-5 5-6 6-7
85.                 pageJumpCounts.filter(data => idZipList.contains(data))
86.             }
87.         ).map(_._2)
88.  
89.         // pageFlowRDD.foreach(println)
90.  
91.         //6.聚合統計結果：(pageId1-pageId2, sum)
92.         val pageFlowMapRDD: RDD[(String, Long)] = pageFlowRDD.flatMap(list => list).map((_, 1L)).reduceByKey(_ + _)
93.  
94.         //7 計算頁面單跳轉換率
95.         pageFlowMapRDD.foreach {
96.             case (pageflow, sum) => {
97.                 val pageIds: Array[String] = pageflow.split("-")
98.                 val pageIdSum: Long = idsMap.getOrElse(pageIds(0).toLong, 1L)
99.  
100.                 println(pageflow + "=" + sum.toDouble / pageIdSum)
101.             }
102.         }
103.  
104.         //8.關閉連接
105.         sc.stop()
106.     }
107. }