Spark使用RDD实现分组topN(八种方法)

最近在复习Spark，记录一个使用RDD实现分组topN的方法，一共写了八种，其中有很多地方都是有共性的，我会在代码最后进行总结八种的思路，他们之间的共性以及每一种的优缺点。

以下是样例数据

语文,赵三金
语文,赵三金
语文,赵三金
语文,赵三金
语文,赵三金
语文,赵三金
语文,杀马特团长
语文,杀马特团长
语文,杀马特团长
语文,杀马特团长
语文,杀马特团长
语文,杀马特团长
语文,杀马特团长
语文,杀马特团长
语文,杀马特团长
语文,杀马特团长
语文,杀马特团长
语文,杀马特团长
语文,杀马特团长
语文,杀马特团长
语文,杀马特团长
数学,黑牛
数学,黑牛
数学,黑牛
数学,黑牛
数学,黑牛
数学,黑牛
数学,白牛
数学,白牛
数学,白牛
数学,白牛
数学,白牛
数学,白牛
数学,白牛
数学,刀哥
数学,刀哥
数学,刀哥
数学,刀哥
数学,刀哥
数学,刀哥
数学,刀哥
数学,刀哥
数学,刀哥
数学,虎哥
数学,虎哥
数学,虎哥
数学,虎哥
数学,虎哥
数学,虎哥
数学,虎哥
数学,虎哥
数学,虎哥
数学,虎哥
数学,虎哥
数学,虎哥
数学,虎哥
数学,虎哥
数学,虎哥
数学,虎哥
数学,虎哥
数学,虎哥
数学,虎哥

-----求每个学科最受欢迎的老师前两名

import org.apache.spark.{SparkConf, SparkContext}/*** 多种思路取分组topN 例如 每个学科最受欢迎的老师* 第一种：* 先统计每个学科每个老师的访问次数* 然后根据学科分组 统计每个老师访问次数前N次的* 这里的例子N取2*/
object topNDemo1 {def main(args: Array[String]): Unit = {val conf = new SparkConf().setMaster("local[*]").setAppName("Demo1")val sc = new SparkContext(conf)//1.首先读入文件,切割,并将二元组和1组合成三元组，这样就能统计每个老师的次数了val rdd1 = sc.textFile("data/topN.txt")val rdd2 = rdd1.mapPartitions(it => {it.map(x => {val strings = x.split(",")((strings(0), strings(1)), 1)})})/*** rdd2的示例数据* ((数学,虎哥),1)* ((语文,杀马特团长),1)* ((数学,虎哥),1)*///2.通过reduceByKey统计出每个学科每个老师访问次数val rdd3 = rdd2.reduceByKey(_ + _)/*** rdd3的数据* ((语文,杀马特团长),15)* ((语文,赵三金),6)* ((数学,虎哥),19)* ((数学,刀哥),9)* ((数学,黑牛),6)* ((数学,白牛),7)*///3.根据学科分组,对value toList后排序取前N个val rdd4 = rdd3.groupBy(_._1._1)/*** rdd4的数据* (数学,CompactBuffer(((数学,刀哥),9), ((数学,虎哥),19), ((数学,黑牛),6), ((数学,白牛),7)))* (语文,CompactBuffer(((语文,赵三金),6), ((语文,杀马特团长),15)))*/val rdd5 = rdd4.flatMapValues(it => {//根据访问次数降序排列,取出N个。注意：这里toList的时候容易造成OOMit.toList.sortBy(-_._2).take(2)})/*** rdd5的数据* (数学,((数学,虎哥),19))* (数学,((数学,刀哥),9))* (语文,((语文,杀马特团长),15))* (语文,((语文,赵三金),6))*///4.最后map取出要的数据即可val result = rdd5.map(x => {(x._1, x._2._1._2, x._2._2)})/*** 结果数据* (数学,虎哥,19)* (数学,刀哥,9)* (语文,杀马特团长,15)* (语文,赵三金,6)*/result.foreach(println)sc.stop()}
}

import org.apache.spark.{SparkConf, SparkContext}/*** 多种思路取分组topN 例如 每个学科最受欢迎的老师* 第二种：* 先统计每个学科每个老师的访问次数* 然后拿出一个学科,或者for循环出每一个学科,使用top方法取出前N个*/
object topNDemo2 {def main(args: Array[String]): Unit = {val conf = new SparkConf().setMaster("local[*]").setAppName("Demo1")val sc = new SparkContext(conf)//1.首先读入文件,切割,并将二元组和1组合成三元组，这样就能统计每个老师的次数了val rdd1 = sc.textFile("data/topN.txt")val rdd2 = rdd1.mapPartitions(it => {it.map(x => {val strings = x.split(",")((strings(0), strings(1)), 1)})})/*** rdd2的示例数据* ((数学,虎哥),1)* ((语文,杀马特团长),1)* ((数学,虎哥),1)*///2.通过reduceByKey统计出每个学科每个老师访问次数val rdd3 = rdd2.reduceByKey(_ + _)/*** rdd3的数据* ((语文,杀马特团长),15)* ((语文,赵三金),6)* ((数学,虎哥),19)* ((数学,刀哥),9)* ((数学,黑牛),6)* ((数学,白牛),7)*///3.过滤要统计的学科,如果统计的科目是固定的可以写一个for循环val filtered = rdd3.filter(_._1._1.equals("数学"))//4.再调用top方法，根据传入的排序规则拿出前两个即可/*** def top(num: Int)(implicit ord: Ordering[T]): Array[T] = withScope {* takeOrdered(num)(ord.reverse)* }* top的源码将我们传入的排序顺序反转了，这里要注意*/implicit val orderRule = Ordering[Int].on[((String, String), Int)](_._2)val result = filtered.top(2)result.foreach(println)sc.stop()}
}

import org.apache.spark.{Partitioner, SparkConf, SparkContext}import scala.collection.mutable/*** 多种思路取分组topN 例如 每个学科最受欢迎的老师* 第三种：* 自定义分区器,将一个学科的数据放到用一个分区内,再进行toList sort 取前N个*/
object topNDemo3 {def main(args: Array[String]): Unit = {val conf = new SparkConf().setMaster("local[*]").setAppName("Demo1")val sc = new SparkContext(conf)//1.首先读入文件,切割,并将二元组和1组合成三元组，这样就能统计每个老师的次数了val rdd1 = sc.textFile("data/topN.txt")val rdd2 = rdd1.mapPartitions(it => {it.map(x => {val strings = x.split(",")((strings(0), strings(1)), 1)})})/*** rdd2的示例数据* ((数学,虎哥),1)* ((语文,杀马特团长),1)* ((数学,虎哥),1)*///2.通过reduceByKey统计出每个学科每个老师访问次数val rdd3 = rdd2.reduceByKey(_ + _)/*** rdd3的数据* ((语文,杀马特团长),15)* ((语文,赵三金),6)* ((数学,虎哥),19)* ((数学,刀哥),9)* ((数学,黑牛),6)* ((数学,白牛),7)*///3.根据自己定义的分区器将数据传入不同的分区,这里要先处理一下判断有多少学科val subjects = rdd3.map(_._1._1).distinct().collect()val partitioner = new MyPartition(subjects)val rdd4 = rdd3.partitionBy(partitioner)//4.再对每个分区内的数据排序取前两个val result = rdd4.mapPartitions(it => {it.toList.sortBy(-_._2).take(2).iterator})result.foreach(println)sc.stop()}
}class MyPartition(subjects: Array[String]) extends Partitioner {val subToNum = new mutable.HashMap[String, Int]var i = 0for (elem <- subjects) {subToNum(elem) = ii += 1}//判断分区的个数override def numPartitions: Int = subjects.size//数据进入分区的规则override def getPartition(key: Any): Int = {val tp = key.asInstanceOf[(String, String)]subToNum(tp._1)}
}

import org.apache.spark.{Partitioner, SparkConf, SparkContext}import scala.collection.mutable/*** 多种思路取分组topN 例如 每个学科最受欢迎的老师* 第四种：* 自定义分区器,将一个学科的数据放到用一个分区内,使用TreeSet保存数据,这样数据量再大也不会OOM*/
object topNDemo4 {def main(args: Array[String]): Unit = {val conf = new SparkConf().setMaster("local[*]").setAppName("Demo1")val sc = new SparkContext(conf)//1.首先读入文件,切割,并将二元组和1组合成三元组，这样就能统计每个老师的次数了val rdd1 = sc.textFile("data/topN.txt")val rdd2 = rdd1.mapPartitions(it => {it.map(x => {val strings = x.split(",")((strings(0), strings(1)), 1)})})/*** rdd2的示例数据* ((数学,虎哥),1)* ((语文,杀马特团长),1)* ((数学,虎哥),1)*///2.通过reduceByKey统计出每个学科每个老师访问次数val rdd3 = rdd2.reduceByKey(_ + _)/*** rdd3的数据* ((语文,杀马特团长),15)* ((语文,赵三金),6)* ((数学,虎哥),19)* ((数学,刀哥),9)* ((数学,黑牛),6)* ((数学,白牛),7)*///3.根据自己定义的分区器将数据传入不同的分区,这里要先处理一下判断有多少学科val subjects = rdd3.map(_._1._1).distinct().collect()val partitioner = new MyPartition(subjects)val rdd4 = rdd3.partitionBy(partitioner)rdd4.foreach(println)//4.再对每个分区内的数据遍历,放入TreeSet中val result = rdd4.mapPartitions(it => {//隐式传入一个排序规则implicit val rules = Ordering[Int].on[((String, String), Int)](-_._2)val sorter = new mutable.TreeSet[((String, String), Int)]()it.foreach(x => { // 这里千万不能调用map,不然会返回一个空的TreeSet//取前N个的话,TreeSet里存N+1个数就可以sorter += xif (sorter.size > 2) {sorter -= sorter.last}})sorter.iterator})result.foreach(println)sc.stop()}
}

import org.apache.spark.{SparkConf, SparkContext}import scala.collection.mutable/*** 多种思路取分组topN 例如 每个学科最受欢迎的老师* 第五种：* 不使用自定义分区器,在groupBy之后直接放到TreeSet里*/
object topNDemo5 {def main(args: Array[String]): Unit = {val conf = new SparkConf().setMaster("local[*]").setAppName("Demo1")val sc = new SparkContext(conf)//1.首先读入文件,切割,并将二元组和1组合成三元组，这样就能统计每个老师的次数了val rdd1 = sc.textFile("data/topN.txt")val rdd2 = rdd1.mapPartitions(it => {it.map(x => {val strings = x.split(",")((strings(0), strings(1)), 1)})})/*** rdd2的示例数据* ((数学,虎哥),1)* ((语文,杀马特团长),1)* ((数学,虎哥),1)*///2.通过reduceByKey统计出每个学科每个老师访问次数val rdd3 = rdd2.reduceByKey(_ + _)//3.对已经统计好的RDD按照学科进行分组val grouped = rdd3.groupBy(_._1._1)//4.对已经分好组的数据的Value放到一个TreeSet里val sorted = grouped.flatMapValues(it => {//隐式传入一个排序规则implicit val rules = Ordering[Int].on[((String, String), Int)](-_._2)val sorter = new mutable.TreeSet[((String, String), Int)]()it.foreach(x => { // 这里千万不能调用map,不然会返回一个空的TreeSet//取前N个的话,TreeSet里存N+1个数就可以sorter += xif (sorter.size > 2) {sorter -= sorter.last}})sorter.iterator})//5.map一下拿出要的数据val result = sorted.map(x => {x._2})result.foreach(println)sc.stop()}
}

import org.apache.spark.{SparkConf, SparkContext}import scala.collection.mutable/*** 多种思路取分组topN 例如 每个学科最受欢迎的老师* 第六种：* 自定义分区器,将一个学科的数据放到用一个分区内,使用TreeSet保存数据,这样数据量再大也不会OOM* 并且进行优化，可以将分区的过程放到reduceByKey，这样减少shuffle提高效率*/
object topNDemo6 {def main(args: Array[String]): Unit = {val conf = new SparkConf().setMaster("local[*]").setAppName("Demo1")val sc = new SparkContext(conf)//1.首先读入文件,切割,并将二元组和1组合成三元组，这样就能统计每个老师的次数了val rdd1 = sc.textFile("data/topN.txt")val rdd2 = rdd1.mapPartitions(it => {it.map(x => {val strings = x.split(",")((strings(0), strings(1)), 1)})})/*** rdd2的示例数据* ((数学,虎哥),1)* ((语文,杀马特团长),1)* ((数学,虎哥),1)*/val subjects = rdd2.map(_._1._1).distinct().collect()val partitioner = new MyPartition(subjects)//2.通过reduceByKey统计出每个学科每个老师访问次数//3.根据自己定义的分区器将数据传入不同的分区,这里要先处理一下判断有多少学科val rdd3 = rdd2.reduceByKey(partitioner,_ + _)/*** rdd3的数据* ((语文,杀马特团长),15)* ((语文,赵三金),6)* ((数学,虎哥),19)* ((数学,刀哥),9)* ((数学,黑牛),6)* ((数学,白牛),7)*///4.再对每个分区内的数据遍历,放入TreeSet中val result = rdd3.mapPartitions(it => {//隐式传入一个排序规则implicit val rules = Ordering[Int].on[((String, String), Int)](-_._2)val sorter = new mutable.TreeSet[((String, String), Int)]()it.foreach(x => { // 这里千万不能调用map,不然会返回一个空的TreeSet//取前N个的话,TreeSet里存N+1个数就可以sorter += xif (sorter.size > 2) {sorter -= sorter.last}})sorter.iterator})result.foreach(println)sc.stop()}
}

import org.apache.spark.{Partitioner, SparkConf, SparkContext}import scala.collection.mutable/*** 多种思路取分组topN 例如 每个学科最受欢迎的老师* 第七种：* 自定义分区器,将一个学科的数据放到用一个分区内,并且调用repartitionAndSortWithinPartitions方法**/
object topNDemo7 {def main(args: Array[String]): Unit = {val conf = new SparkConf().setMaster("local[*]").setAppName("Demo1")val sc = new SparkContext(conf)//1.首先读入文件,切割,并将二元组和1组合成三元组，这样就能统计每个老师的次数了val rdd1 = sc.textFile("data/topN.txt")val rdd2 = rdd1.mapPartitions(it => {it.map(x => {val strings = x.split(",")((strings(0), strings(1)), 1)})})/*** rdd2的示例数据* ((数学,虎哥),1)* ((语文,杀马特团长),1)* ((数学,虎哥),1)*///2.通过reduceByKey统计出每个学科每个老师访问次数val rdd3 = rdd2.reduceByKey(_ + _)//3.将数据变成这种形式是为了调用下边的算子,该算子只能对key排序val rdd4 = rdd3.map(tp => {((tp._1._1, tp._1._2, tp._2), null)})rdd4/*** rdd3的数据* ((语文,杀马特团长),15)* ((语文,赵三金),6)* ((数学,虎哥),19)* ((数学,刀哥),9)* ((数学,黑牛),6)* ((数学,白牛),7)*/val subjects = rdd3.map(_._1._1).distinct().collect()val partitioner = new MyPartition2(subjects)//3.使用重新分区并且排序的方法,底层调用的是new一个ShuffleRDD,并且隐式传入key的排序规则implicit  val value = Ordering[Int].on[(String, String,Int)](-_._3)val rdd5 = rdd4.repartitionAndSortWithinPartitions(partitioner)//4.这样数据就整体有序了,再取出前N个就可以了val result = rdd5.map(_._1)result.foreach(println)sc.stop()}
}class MyPartition2(subjects: Array[String]) extends Partitioner {val subToNum = new mutable.HashMap[String, Int]var i = 0for (elem <- subjects) {subToNum(elem) = ii += 1}//判断分区的个数override def numPartitions: Int = subjects.size//数据进入分区的规则override def getPartition(key: Any): Int = {val tp = key.asInstanceOf[(String, String,Int)]subToNum(tp._1)}
}

import org.apache.spark.rdd.ShuffledRDD
import org.apache.spark.{Partitioner, SparkConf, SparkContext}import scala.collection.mutable/*** 多种思路取分组topN 例如 每个学科最受欢迎的老师* 第四种：* 自定义分区器,将一个学科的数据放到用一个分区内,使用shuffledRDD运算*/
object topNDemo8 {def main(args: Array[String]): Unit = {val conf = new SparkConf().setMaster("local[*]").setAppName("Demo1")val sc = new SparkContext(conf)//1.首先读入文件,切割,并将二元组和1组合成三元组，这样就能统计每个老师的次数了val rdd1 = sc.textFile("data/topN.txt")val rdd2 = rdd1.mapPartitions(it => {it.map(x => {val strings = x.split(",")((strings(0), strings(1)), 1)})})/*** rdd2的示例数据* ((数学,虎哥),1)* ((语文,杀马特团长),1)* ((数学,虎哥),1)*///2.通过reduceByKey统计出每个学科每个老师访问次数val rdd3 = rdd2.reduceByKey(_ + _)//3.将数据变成这种形式是为了调用下边的算子,该算子只能对key排序val rdd4 = rdd3.map(tp => {((tp._1._1, tp._1._2, tp._2), null)})rdd4/*** rdd3的数据* ((语文,杀马特团长),15)* ((语文,赵三金),6)* ((数学,虎哥),19)* ((数学,刀哥),9)* ((数学,黑牛),6)* ((数学,白牛),7)*/val subjects = rdd3.map(_._1._1).distinct().collect()val partitioner = new MyPartition3(subjects)//3.使用重新分区并且排序的方法,底层调用的是new一个ShuffleRDD,并且隐式传入key的排序规则implicit val value = Ordering[Int].on[(String, String, Int)](_._3)val shuffledRDD = new ShuffledRDD[(String, String, Int), Null, Null](rdd4, partitioner)shuffledRDD.setKeyOrdering(value)shuffledRDD.foreach(println)}
}class MyPartition3(subjects: Array[String]) extends Partitioner {val subToNum = new mutable.HashMap[String, Int]var i = 0for (elem <- subjects) {subToNum(elem) = ii += 1}//判断分区的个数override def numPartitions: Int = subjects.size//数据进入分区的规则override def getPartition(key: Any): Int = {val tp = key.asInstanceOf[(String, String, Int)]subToNum(tp._1)}
}

总结：

第一种方法：最基本方法。先统计每个学科每个老师的访问次数,然后按照学科分组,对所有老师和访问次数转化成list再排序，这种方法如果数据量非常大的话容易造成内存溢出。

第二种方法：使用top。先统计每个学科每个老师的访问次数,然后做一次收集,统计出一共有多少学科，然后for循环遍历每一个学科,然后取出topN就可以了，这里要自定义排序规则，由于top方法是先在分区里取top再在分区间取top，所有不会内存溢出。但是这种方法如果有很多学科，会触发非常多的Action。

第三种方法：使用自定义分区器。先统计每个学科每个老师的访问次数,然后自定义一个分区器，按照学科分区，再对每个分区内的数据转化成list排序，这种方法和第一种方法其实很像,因为groupBy也是重新分区，但是这种方法会比groupBy传输的数据要少，因此要快一点，但是还是有内存溢出的风险。

第四种方法：使用自定义分区器+TreeSet。先统计每个学科每个老师的访问次数,然后自定义一个分区器，按照学科分区，然后对每个分区遍历，拿出里边的数据放到TreeSet里，这里要自定义一个排序规则，如果要取TOP2的话，TreeSet里最多存储两个数据就够了，每次第三个数据存进TreeSet里的时候进行排序然后去掉最后一个，这样的话不会有内存溢出的风险。

第五种方法：直接使用TreeSet存储数据。统计每个学科每个老师的访问次数,然后根据学科分组，再对values进行map，在组内使用TreeSet排序,就和第四种一样了。和第四种的区别就是，第四种按照学科分区后对每个分区一个TreeSet，第五种是按学科分组后对每个组内一个TreeSet。

第六种方法：对第四种进行优化。第四种方法先统计访问次数再自定义分区器,都会产生shuffle，其实可以把分区的过程提前，放到reduceByKey算子里，这样就减少了shuffle的次数，其实影响不是很大，因为在第四种方法里，reduce后产生的结果也会被复用。

第七种方法：调用repartitionAndSortWithinPartitions方法，这个算子可以直接根据我们传入的分区器和排序方法对数据进行分区并且对key进行排序，注意这里只能对key进行排序，因为底层调用的是new ShuffledRDD.setKeyOrdering 方法，所以要注意把排序的数据放到key中。这样到下游数据就是有序的，取出前N个即可。

第八种方法：第七种方法调用的算子底层是ShuffledRDD，因此也可以直接new一个ShuffledRDD来完成需求，这里不多解释。

Spark使用RDD实现分组topN(八种方法)相关推荐

提高IIS网站服务器的效率的八种方法 (转载）
作者:未知请作者速与本人联系以下是提高IIS 5.0网站服务器的执行效率的八种方法: 1. 启用HTTP的持续作用可以改善15~20%的执行效率. 2. 不启用记录可以改善5~8%的执行效率. ...
Linux下进程通信的八种方法
Linux下进程通信的八种方法:管道(pipe),命名管道(FIFO),内存映射(mapped memeory),消息队列(message queue),共享内存(shared memory),信号量 ...
python管道安装包_Python 炫技操作：安装包的八种方法
1. 使用 easy_install easy_install 这应该是最古老的包安装方式了,目前基本没有人使用了.下面是 easy_install 的一些安装示例# 通过包名,从PyPI寻找最新版本 ...
译 | 在 Azure SQL 上节约成本的八种方法
点击上方关注"汪宇杰博客" 原文:John 'JG' Chirapurath 翻译:汪宇杰导语当今世界,企业正崭露头角,渴望重整与重建,但仍处于不确定的时期.节约成本并将开支重 ...
MATLAB求解线性方程组的八种方法
MATLAB求解线性方程组的八种方法求解线性方程分为两种方法–直接法和迭代法常见的方法一共有8种直接法 Gauss消去法 Cholesky分解法迭代法 Jacobi迭代法 Gauss-Seid ...
Python炫技操作：花式导包的八种方法
点击上方"Python爬虫与数据挖掘",进行关注回复"书籍"即可获赠Python从入门到进阶共10本电子书今日鸡汤五岳寻仙不辞远,一身好入名山游. ...
HTTP之原理,长短连接,响应码,三握四挥,八种方法
文章目录 1 HTTP原理 1.1 HTTP协议与TCP/IP协议的关系 1.2 无状态的HTTP协议 2 长短连接详解 2.1 简介 2.2 TCP短连接 2.3 TCP长连接 2.4 长连接短连接 ...
【fraud detection】防网络广告作弊（点击欺诈）的八种方法
防网络广告作弊(点击欺诈)的八种方法网络广告防作弊(点击欺诈)的技术手段有很多种,归纳起来,大致有以下八点: 1. IP防止作弊:一般计费方式是按照24小时内唯一IP,可以将每个IP记入数据库,当下 ...
防网络广告作弊（点击欺诈）的八种方法
防网络广告作弊(点击欺诈)的八种方法网络广告防作弊(点击欺诈)的技术手段有很多种,归纳起来,大致有以下八点: 1. IP防止作弊:一般计费方式是按照24小时内唯一IP,可以将每个IP记入数据库,当下 ...

Spark使用RDD实现分组topN(八种方法)

Spark使用RDD实现分组topN(八种方法)相关推荐

最新文章

热门文章