Spark开发实战-主播打赏排行榜统计

（一）需求分析

计算每个大区当天金币收入排名前N的主播

背景：
我们有一款直播APP，已经在很多国家上线并运营了一段时间，产品经理希望开发一个功能，计算前N主播排行榜，按天更新排名信息，统计的维度有多种，其中有一个维度是针对主播当天直播的金币收入进行排名。
一个大区下面包含多个国家，不同大区的运营策略是不一样的，所以就把不同国家划分到不同大区里面，方便运营。
那这个TopN主播排行榜在统计的时候就需要分大区统计了。
针对主播每天的开播数据我们已经有了，以及直播间内用户的送礼记录也都是有的。那这样其实就可以统计主播当天的金币收入了主播一天可能会开播多次，所以后期在统计主播当天收入的时候是需要把他当天所有直播中的金币收入都计算在内的。

分析 ：我们有两份数据，数据都是json格式的

1. video_info.log 主播的开播记录，其中包含主播的id：uid、直播间id：vid 、大区：area、视频开播时长：length、增加粉丝数量：follow等信息
2. gift_record.log 用户送礼记录，其中包含送礼人id：uid，直播间id：vid，礼物id：good_id，金币数
   量：gold 等信息.

其实就是按照当天主播所有开播的直播间内的收入汇总，按大区分组，统计每个大区内收入排名前N的主播。

（二）开发步骤

1：首先获取两份数据中的核心字段，使用fastjson包解析数据
主播开播记录：主播ID：uid，直播间ID：vid，大区：area
(vid,(uid,area))用户送礼记录：直播间ID：vid，金币数量：gold(vid,gold) 这样的可以把这两份数据关联到一块就能获取到大区、主播、金币这些信息了，使用直播间vid进行关联。

2：对用户送礼记录数据进行聚合，对相同vid的数据求和
因为用户可能在一次直播中给主播送多次礼物
(vid,gold_sum)
3：把这两份数据join到一块，vid作为join的key
(vid,((uid,area),gold_sum))
4：使用map迭代join之后的数据，最后获取到uid,area,gold_sum字段，由于一个主播一天可能会开播多
次，后面需要基于uid和area再做一次聚合，所以把数据转换成这种格式uid和area是一一对应的，一个人只能属于一个大区
((uid,area),gold_sum)
5：使用reduceByKey算子对数据进行聚合
((uid,area),gold_sum_all)
6：接下来对需要使用groupByKey对数据进行分组，所以先使用map进行转换
map：(area,(uid,gold_sum_all))
groupByKey: area,<(uid,gold_sum_all),(uid,gold_sum_all),(uid,gold_sum_all)>
7：使用map迭代每个分组内的数据，按金币数量倒序排序，取前N个，最终输出area、topN
这个TopN其实就是把前几名主播的id还有金币数量拼接成一个字符串(area,topN)
8：使用foreach将结果打印到控制台，多个字段使用制表符分割area topN

（三）环境依赖

<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.11</artifactId>
<version>2.4.3</version>
<!--<scope>provided</scope>-->
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.68</version>
</dependency>

（四）代码开发

object TopNScala {def main(args: Array[String]): Unit = {val conf = new SparkConf()conf.setAppName("TopNScala").setMaster("local")val sc = new SparkContext(conf)//1：首先获取两份数据中的核心字段，使用fastjson包解析数据val videoInfoRDD = sc.textFile("D:\\video_info.log")val giftRecordRDD = sc.textFile("D:\\gift_record.log")//(vid,(uid,area))val videoInfoFieldRDD = videoInfoRDD.map(line=>{val jsonObj = JSON.parseObject(line)val vid = jsonObj.getString("vid")val uid = jsonObj.getString("uid")val area = jsonObj.getString("area")(vid,(uid,area))})//(vid,gold)val giftRecordFieldRDD = giftRecordRDD.map(line=>{val jsonObj = JSON.parseObject(line)val vid = jsonObj.getString("vid")val gold = Integer.parseInt(jsonObj.getString("gold"))(vid,gold)})//2：对用户送礼记录数据进行聚合，对相同vid的数据求和//(vid,gold_sum)val giftRecordFieldAggRDD = giftRecordFieldRDD.reduceByKey(_ + _)//3：把这两份数据join到一块，vid作为join的key//(vid,((uid,area),gold_sum))val joinRDD = videoInfoFieldRDD.join(giftRecordFieldAggRDD)//4：使用map迭代join之后的数据，最后获取到uid、area、gold_sum字段//joinRDD: (vid,((uid,area),gold_sum))val joinMapRDD =joinRDD.map(tup=>{//joinRDD: (vid,((uid,area),gold_sum))//获取uidval uid = tup._2._1._1//获取areaval area = tup._2._1._2//获取gold_sumval gold_sum = tup._2._2((uid,area),gold_sum)})//5：使用reduceByKey算子对数据进行聚合//((uid,area),gold_sum_all)val reduceRDD = joinMapRDD.reduceByKey(_ + _)//6：接下来对需要使用groupByKey对数据进行分组，所以先使用map进行转换//map：(area,(uid,gold_sum_all))//groupByKey: area,<(uid,gold_sum_all),(uid,gold_sum_all),(uid,gold_sum_all)val groupRDD = reduceRDD.map(tup=>(tup._1._2,(tup._1._1,tup._2))).groupByKey()//7：使用map迭代每个分组内的数据，按照金币数量倒序排序，取前N个，最终输出area,t//(area,topN)val top3RDD = groupRDD.map(tup=>{val area = tup._1//toList：把iterable转成list//sortBy：排序，默认是正序//reverse：反转，实现倒序效果//take(3)：取前3个元素//mkString：使用指定字符把集合转成字符串//uid:gold_sum_all,uid:gold_sum_all,uid:gold_sum_allval top3 = tup._2.toList.sortBy(_._2).reverse.take(3).map(tup=>tup._1+":"+tup._2).mkString(",")(area,top3)})//8：使用foreach将结果打印到控制台，多个字段使用制表符分割top3RDD.foreach(tup=>println(tup._1+"\t"+tup._2))sc.stop()}
}