Spark取出(Key,Value)型数据中Value值为前n条数据

最近在使用Spark进行一些日志分析，需要对日志中的一些(key,value)型数据进行排序，并取出value最多的10条数据。经过查找资料，发现Spark中的top()函数可以取出排名前n的元素，以及sortBy()函数可以对(key,value)数据根据value进行排序，原以为一切都很好解决，但是实际情况并没有得到想要的结果数据，研究了部分源码，才最终达到了想要的数据，特在此备注和分享。

前期遇到的坑

刚开始，通过查找资料，知道Spark可以使用sortByKey()和sortBy() 两个函数对(key,value)型数据排序。于是，直接使用sortByKey()进行排序，排完之后才发现，排序的时候是根据Key排序，而我需要先对Key进行汇总，再根据Value进行排序。显然，sortByKey不能满足需求！

于是，开始尝试使用sortBy()函数，使用方法为 rdd.sortBy(_._2,false)，即可对value进行降序排序。在测试的时候，我使用了rdd.sortBy(_._2,false).collect()进行排序和汇总，但是collect()函数会将所有的数据汇总到Driver，当数据量太大时对导致Driver中的内存不足。于是，想着只取将10条数据返回给Driver。经过查找，知道top()函数可以取出前10条数据。

top()函数中的坑及其解决方法

知道可以用top()函数取出前10条数据，以为这么简单就能得到想要的数据，好激动！谁知，我还是高兴得太早了-_-

我取出Value值排名前10的数据的实现代码为: rdd.sortBy(_._2,false).top(10) 代码简洁清晰，感觉一切都是这么美好。然而，当我放到集群上运行时，得到的结果却大大出乎我的意料。Value值为1的数据都被取出来了，Value值较大的数据反而没有取到，神马情况？？

想了半天，也不知道是哪里出了问题。没办法，只能研究源码了。于是，花了点时间研究将sortBy()函数的源码看了一遍，发现sortBy()函数的实现其实最终是调用了sortByKey()函数进行排序。但是，这跟最后取得Value值为前10的目标没有冲突呀！于是，继续研究top()函数的源码。由于之前一直想着top()函数就是取出排序之后的前n条数据，已经有这个惯性思维一直在脑海中，所以在前两次看top()源码的时候，也没发现什么异常。下面为top()函数的源码实现：

后来，静下心来再看了几遍，终于发现不是太对了。我的数据类型是(key,value)格式的，rdd.sortBy(_._2,false)中实现了根据value值排序的目的，但是 .top(10) 却取出了key为前10的数据。top()函数源码中，对RDD中的数据进行了reduce操作，并将结果进行排序。所以，rdd.sortBy(_._2,false).top(10) 这段代码先是对(key,value)数据根据value进行排序，而top()函数中，数据又再次对key进行了排序，导致之前根绝value排序的结果乱序了，所以最后取到的是key排在前10的数据。这就是导致问题的原因，终于被我发现了！

问题虽然被发现了，但是怎么解决呢？说实话，我对Scala也不是太了解，只能去QQ群里请教了一些大神。有一位叫做老徐的大神帮我给出了解决方法： rdd.sortBy(_._2,false).top(10)(Ordering.by(e => e._2))。再次运行，果然能得到正确结果。后来再仔细想想，觉得sortBy()函数有点多余，于是变成rdd.top(10)(Ordering.by(e => e._2))。至此，已经能对(key,value)类型的数据进行汇总，然后根据value值进行排序，最后取出value排名前10的数据了。

take()函数实现目标

在请教大神的时候，偶然接触到了take()函数，经过测试: rdd.sortBy(_._2,false).take(10) 这段代码能得到value排名前10的数据。

查看take()函数的源码，如下：

[java] view plaincopy

/**
* Take the first num elements of the RDD. It works by first scanning one partition, and use the
* results from that partition to estimate the number of additional partitions needed to satisfy
* the limit.
*
* @note this method should only be used if the resulting array is expected to be small, as
* all the data is loaded into the driver's memory.
*
* @note due to complications in the internal implementation, this method will raise
* an exception if called on an RDD of `Nothing` or `Null`.
*/
def take(num: Int): Array[T] = withScope {
if (num == 0) {
new Array[T](0)
} else {
val buf = new ArrayBuffer[T]
val totalParts = this.partitions.length
var partsScanned = 0
while (buf.size < num && partsScanned < totalParts) {
// The number of partitions to try in this iteration. It is ok for this number to be
// greater than totalParts because we actually cap it at totalParts in runJob.
var numPartsToTry = 1L
if (partsScanned > 0) {
// If we didn't find any rows after the previous iteration, quadruple and retry.
// Otherwise, interpolate the number of partitions we need to try, but overestimate
// it by 50%. We also cap the estimation in the end.
if (buf.size == 0) {
numPartsToTry = partsScanned * 4
} else {
// the left side of max is >=1 whenever partsScanned >= 2
numPartsToTry = Math.max((1.5 * num * partsScanned / buf.size).toInt - partsScanned, 1)
numPartsToTry = Math.min(numPartsToTry, partsScanned * 4)
}
}
val left = num - buf.size
val p = partsScanned.until(math.min(partsScanned + numPartsToTry, totalParts).toInt)
val res = sc.runJob(this, (it: Iterator[T]) => it.take(left).toArray, p)
res.foreach(buf ++= _.take(num - buf.size))
partsScanned += p.size
}
buf.toArray
}
}

该函数的注解指出，take()函数通过扫描一个数据分区，并取出该分区中的前n个数据，避免了其它分区数据的检索。最主要的是，该函数没有对父RDD中的数据进行重新分区，所以，数据的分区和排序顺序并没有改变，因此能取出value排名前10的数据。

总结

经过上面的这些折腾，发现top()函数中所遇到的坑的实质是由于(key,value)数据在sortBy(_._2)函数中根据value进行排序的时候，会进行Shuffle操作，根据value值将原来的数据进行重新分区。而sortBy()对数据排序之后，在top()函数中进行排序时，会根据key进行Shuffle操作，并得到根据key排序和分区之后的新RDD，所以导致最后的结果跟预期的不一致。

只要肯花时间，自己的潜力还是可以被挖掘出来的！