| spark 对于key/value 排序 取前几 top take(转载) | 您所在的位置：网站首页 › gbase取前十条 › spark 对于key/value 排序 取前几 top take(转载) |
spark 对于key/value 排序 取前几 top take(转载)
|
看到以前很细的关于排序的文章，偷偷的将其收入希望可以让更多人看到。 来源：http://blog.csdn.net/dai451954706/article/details/52668258
最近在使用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 plain copy Flash Content! Click to Activate Flash /** * 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，所以导致最后的结果跟预期的不一致。 |
【本文地址】