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()函数中的坑及其解决方法
后来,静下心来再看了几遍,终于发现不是太对了。我的数据类型是(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()函数的源码,如下:
- /**
- * 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
- }
- }
总结
经过上面的这些折腾,发现top()函数中所遇到的坑的实质是由于(key,value)数据在sortBy(_._2)函数中根据value进行排序的时候,会进行Shuffle操作,根据value值将原来的数据进行重新分区。而sortBy()对数据排序之后,在top()函数中进行排序时,会根据key进行Shuffle操作,并得到根据key排序和分区之后的新RDD,所以导致最后的结果跟预期的不一致。
只要肯花时间,自己的潜力还是可以被挖掘出来的!