介绍本章要点spark api.pptxVIP

sparkAPI(二)

本章要点一、概述二、sortBy函数实现以及使用三、sortByKey函数实现以及使用

概述在很多应用场景都需要对结果数据进行排序，Spark中有时也不例外。在Spark中存在两种对RDD进行排序的函数，分别是sortBy和sortByKey函数。sortBy是对标准的RDD进行排序，它是从Spark0.9.0之后才引入的（可以参见SPARK-1063）。而sortByKey函数是对PairRDD进行排序，也就是有Key和Value的RDD。下面将分别对这两个函数的实现以及使用进行说明。

sortBy函数实现以及使用sortBy函数是在org.apache.spark.rdd.RDD类中实现的，它的实现如下defsortBy[K](f:(T)=K,ascending:Boolean=true,numPartitions:Int=this.partitions.size)(implicitord:Ordering[K],ctag:ClassTag[K]):RDD[T]=this.keyBy[K](f).sortByKey(ascending,numPartitions).values

sortBy函数实现以及使用该函数最多可以传三个参数：第一个参数是一个函数，该函数的也有一个带T泛型的参数，返回类型和RDD中元素的类型是一致的；第二个参数是ascending，从字面的意思大家应该可以猜到，是的，这参数决定排序后RDD中的元素是升序还是降序，默认是true，也就是升序；第三个参数是numPartitions，该参数决定排序后的RDD的分区个数，默认排序后的分区个数和排序之前的个数相等，即为this.partitions.size。

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

使用scalavaldata=List(3,1,90,3,5,12)data:List[Int]=List(3,1,90,3,5,12)scalavalrdd=sc.parallelize(data)rdd:org.apache.spark.rdd.RDD[Int]=ParallelCollectionRDD[0]atparallelizeatconsole:14scalardd.collectres0:Array[Int]=Array(3,1,90,3,5,12)scalardd.sortBy(x=x).collectres1:Array[Int]=Array(1,3,3,5,12,90)

使用scalardd.sortBy(x=x,false).collectres3:Array[Int]=Array(90,12,5,3,3,1)scalavalresult=rdd.sortBy(x=x,false)result:org.apache.spark.rdd.RDD[Int]=MappedRDD[23]atsortByatconsole:16scalaresult.partitions.sizeres9:Int=2scalavalresult=rdd.sortBy(x=x,false,1)result:org.apache.spark.rdd.RDD[Int]=MappedRDD[26]atsortByatconsole:16

使用scalaresult.partitions.sizeres10:Int=1上面的实例对rdd中的元素进行升序排序。并对排序后的RDD的分区个数进行了修改，上面的result就是排序后的RDD，默认的分区个数是2，而我们对它进行了修改，所以最后变成了1。

sortByKey函数实现以及使用sortByKey函数作用于Key-Value形式的RDD，并对Key进行排序。它是在org.apache.spark.rdd.OrderedRDDFunc