如何在Scala Spark中对RDD进行排序？

Question

阅读Spark method sortByKey：

sortByKey([ascending], [numTasks])   When called on a dataset of (K, V) pairs where K implements Ordered, returns a dataset of (K, V) pairs sorted by keys in ascending or descending order, as specified in the boolean ascending argument.

是否可能只返回“N”个数量的结果。因此，与其返回所有结果，不如返回前10名。我可以将已排序的集合转换为数组，并使用take方法，但既然这是一个O（N）操作，有没有更有效的方法？

Answer 1

另一个选项（至少来自PySpark 1.2.0）是使用takeOrdered。

按升序排列：

rdd.takeOrdered(10)

按降序排列：

rdd.takeOrdered(10, lambda x: -x)

k、v对的最大k值：

rdd.takeOrdered(10, lambda (k, v): -v)

Answer 2

很可能您已经仔细阅读了源代码：

  class OrderedRDDFunctions {
   // <snip>
  def sortByKey(ascending: Boolean = true, numPartitions: Int = self.partitions.size): RDD[P] = {
    val part = new RangePartitioner(numPartitions, self, ascending)
    val shuffled = new ShuffledRDD[K, V, P](self, part)
    shuffled.mapPartitions(iter => {
      val buf = iter.toArray
      if (ascending) {
        buf.sortWith((x, y) => x._1 < y._1).iterator
      } else {
        buf.sortWith((x, y) => x._1 > y._1).iterator
      }
    }, preservesPartitioning = true)
  }

正如您所说，整个数据必须经过洗牌阶段——如片段中所示。

然而，您对随后调用take（K）的担心可能并不那么准确。此操作不会在所有N项中循环：

  /**
   * 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.
   */
  def take(num: Int): Array[T] = {

这样看来：

O（myRdd.take（K））

Answer 3

如果你只需要前10名，可以使用rdd。前（10）。它避免了排序，因此速度更快。

rdd。top对数据进行一次并行传递，收集堆中每个分区的前N个，然后合并堆。这是一个O（rdd.count）操作。排序将是O（rdd.count log rdd.count），并且会导致大量数据传输-