Saprk排序

1、基础排序算子sortBy和sortByKey

在Spark中存在两种对RDD进行排序的函数，分别是 sortBy和sortByKey函数。sortBy是对标准的RDD进行排序，它是从Spark0.9.0之后才引入的。而sortByKey函数是对PairRDD进行排序，也就是有Key和Value的RDD。下面将分别对这两个函数的实现以及使用进行说明。

1.1 sortBy

sortBy是在RDD下的

该函数最多可以传三个参数：第一个参数是一个函数，该函数的也有一个带T泛型的参数，返回类型和RDD中元素的类型是一致的；第二个参数是ascending，这参数决定排序后RDD中的元素是升序还是降序，默认是true，也就是升序；第三个参数是numPartitions，该参数决定排序后的RDD的分区个数，默认排序后的分区个数和排序之前的个数相等，即为this.partitions.size。

从sortBy函数的实现可以看出，第一个参数是必须传入的，而后面的两个参数可以不传入。而且sortBy函数函数的实现依赖于sortByKey函数，关于sortByKey函数后面会进行说明。keyBy函数也是RDD类中进行实现的，它的主要作用就是将传进来的每个元素作用于f(x)中，并返回tuples类型的元素，也就变成了Key-Value类型的RDD了

1.2 sortByKey

sortByKey是在OrderedRDDFunctions类下的。sortByKey函数作用于Key-Value形式的RDD，并对Key进行排序

该函数返回的RDD一定是ShuffledRDD类型的，因为对源RDD进行排序，必须进行Shuffle操作，而Shuffle操作的结果RDD就是ShuffledRDD。其实这个函数的实现很优雅，里面用到了RangePartitioner，它可以使得相应的范围Key数据分到同一个partition中，然后内部用到了mapPartitions对每个partition中的数据进行排序，而每个partition中数据的排序用到了标准的sort机制，避免了大量数据的shuffle。

在转换调用sortByKey方法时，会从上下文中提取Ordering[K]，private val ordering = implicitly[Ordering[K]]其中implicit def rddToOrderedRDDFunctions[K : Ordering : ClassTag, V: ClassTag](rdd: RDD[(K, V)])这里的k:Ordering意思是指k必须是可以转换成ordering的子类。下面有2种方式

//1、定义K的排序时使用隐式值
implicit val ord: Ordering[Person] = new Ordering[Person] {
override def compare(x: Person, y: Person): Int = { 
x.name.compareTo(y.name)
}}sc.textFile("").foreachRDD(rdd => rdd.map(msg => (Person(msg), msg)).sortByKey())//2、key类实现Ordered[k]接口中的compare方法
case class Person(name:String) extends Ordered[Person]{override def compare(that: Person): Int = {this.name.compare(that.name)} def run(): Unit ={println("Person...")}
}sc.textFile("").foreachRDD(rdd => rdd.map(msg => (Person(msg), msg)).sortByKey())

2、二次排序

二次排序就是指排序的时候考虑2个维度。如我们排序的时候按照第一个列降序排序，有一种情况，第一列的Key相同怎么排，这个时候可能就需要借助二次排序考虑第二列。

比较明智的方法是自定义排序的key，不采用其他的方式是因为现在二次排序，后面可能3次，5次等，采用自定义key的方式只要重新复写自定义的key，就能用sortByKey。

spark要实现比较，就要实现Orderd这个排序的接口，另外一般也会序列化Serializable。sortByKey会根据Key进行排序，但是如果二次排序的话sortByKey不知道key是什么构建的想法就是，基于已有的数据自定义二次排序的key，sortByKey基于这个自定义的key进行比较。我们用mapToPair重新构造内容，加了自定义的key，value的内容就是已有的内容，根据排序然后把自定义的key去掉。

package com.quinto.sort/*** 自定义二次排序的key*/
class SecondarySortKey(val first:Int,val second:Int) extends Ordered[SecondarySortKey] with Serializable {override def compare(that: SecondarySortKey): Int = {if(this.first - that.first!=0){this.first - that.first}else{this.second - that.second}}
}

package com.quinto.sortimport org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}object SecondarySortApp {def main(args: Array[String]): Unit = {val conf = new SparkConf().setAppName("SecondarySort").setMaster("local[*]")val sc = new SparkContext(conf)val lines = sc.textFile("E:\\test\\a.txt")val sorted = lines.map(lines=> ( new SecondarySortKey(lines.split(" ")(0).toInt,lines.split(" ")(1).toInt),lines))sorted.sortByKey(false).map(line => line._2).collect().foreach(println)}
}

3、TopN

3.1 基础TopN

排序后直接使用take算子取出前几个，take算子后返回的是数组，不是rdd，不能用collect。

package com.quinto.sortimport org.apache.spark.{SparkConf, SparkContext}object BasicYopNApp {def main(args: Array[String]): Unit = {val conf = new SparkConf().setAppName("BasicYopNApp").setMaster("local[*]")val sc = new SparkContext(conf)val lines = sc.textFile("E:\\test\\a.txt")//①生成k-v键值对方便sortByKey进行排序（Int已经实现了排序比较的接口）②降序排序③过滤出排序内容本身⑤获取排名前5放入元素内容，元素内容构成一个Arraylines.map(line=>(line.toInt,line)).sortByKey(false).map(line=>line._2).take(5).foreach(println)}
}

3.2 分组TopN

分组排序就是有不同类型的数据，不同数据中每一种类型数据里面的TopN。

object GroupSortTopN {def main(args: Array[String]): Unit = {val conf = new SparkConf().setAppName(s"${GroupSortTopN.getClass.getSimpleName}").setMaster("local[*]")val sc = new SparkContext(conf)val lines = sc.textFile("file:/E:/data/test.txt")//按照科目进行排序val course2Info:RDD[(String, String)] = lines.map(line => {val spaceIndex = line.indexOf(" ")val course = line.substring(0, spaceIndex)val info = line.substring(spaceIndex + 1)(course, info)})//按照科目排序，指的是科目内排序，不是科目间的排序，所以需要把每个科目的信息汇总val course2Infos:RDD[(String, Iterable[String])] = course2Info.groupByKey()//按照key进行分组//分组内的排序val sorted:RDD[(String, mutable.TreeSet[String])] = course2Infos.map{case (course, infos) => {val topN = mutable.TreeSet[String]()(new Ordering[String](){override def compare(x: String, y: String) = {val xScore = x.split("\\s+")(1)val yScore = y.split("\\s+")(1)yScore.compareTo(xScore)}})for(info <- infos) {topN.add(info)}(course, topN.take(5))}}sorted.foreach(println)sc.stop()}
}

3.3 优化分组TopN

上述在编码过程当中使用groupByKey，我们说着这个算子的性能很差，因为没有本地预聚合，所以应该在开发过程当中尽量避免使用，能用其它代替就代替。使用combineByKey模拟

object GroupSortByCombineByKeyTopN {def main(args: Array[String]): Unit = {val conf = new SparkConf().setAppName(s"${GroupSortByCombineByKeyTopN.getClass.getSimpleName}").setMaster("local[*]")val sc = new SparkContext(conf)val lines = sc.textFile("file:/E:/data/test.txt")//按照科目进行排序val course2Info:RDD[(String, String)] = lines.map(line => {val spaceIndex = line.indexOf(" ")val course = line.substring(0, spaceIndex)val info = line.substring(spaceIndex + 1)(course, info)})//按照科目排序，指的是科目内排序，不是科目间的排序，所以需要把每个科目的信息汇总val sorted= course2Info.combineByKey(createCombiner, mergeValue, mergeCombiners)sorted.foreach(println)sc.stop()}def createCombiner(info:String): mutable.TreeSet[String] = {val ts = new mutable.TreeSet[String]()(new Ordering[String](){override def compare(x: String, y: String) = {val xScore = x.split("\\s+")(1)val yScore = y.split("\\s+")(1)yScore.compareTo(xScore)}})ts.add(info)ts}def mergeValue(ab:mutable.TreeSet[String], info:String): mutable.TreeSet[String] = {ab.add(info)if(ab.size > 5) {ab.take(5)} else {ab}}def mergeCombiners(ab:mutable.TreeSet[String], ab1: mutable.TreeSet[String]): mutable.TreeSet[String] = {for (info <- ab1) {ab.add(info)}if(ab.size > 5) {ab.take(5)} else {ab}}
}

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/473670.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！