Spark排序RDD并加入他们的行列答案

【问题标题】：Spark sort RDD and join their rankSpark排序RDD并加入他们的行列
【发布时间】：2015-05-04 10:28:00
【问题描述】：

我有一个RDD[(VertexId, Double)]，我想按_._2 对其进行排序，并用这个RDD 加入索引（排名）。因此，我可以通过filter 获得一个元素及其排名。

目前我按sortBy 对RDD 进行排序，但我不知道如何将RDD 加入其排名。所以我把它作为一个序列收集起来，并用它的索引压缩它。但这不是有效的。我想知道是否有更优雅的方式来做到这一点。

我现在使用的代码是：

val tmpRes = graph.vertices.sortBy(_._2, ascending = false) // Sort all nodes by its PR score in descending order
      .collect() // collect to master, this may be very expensive

    tmpRes.zip(tmpRes.indices) // zip with index

【问题讨论】：

标签： scala apache-spark rdd

【解决方案1】：

如果有任何机会，您只想将 n 个第一个元组带回驱动程序，那么也许您可以使用 takeOrdered(n, [ordering]) 其中 n 是要返回的结果数，并且 ordering 您要使用的比较器。

否则，您可以使用 zipWithIndex 转换将您的RDD[(VertexId, Double)] 转换为具有适当索引的RDD[((VertexId, Double), Long)]（当然您应该在排序后这样做）。

例如：

scala> val data = sc.parallelize(List(("A", 1), ("B", 2)))
scala> val sorted = data.sortBy(_._2)
scala> sorted.zipWithIndex.collect()
res1: Array[((String, Int), Long)] = Array(((A,1),0), ((B,2),1))

问候，

【讨论】：

谢谢 ssaboum。我以为zipWithIndex 会压缩 index_in_partition + partition_offset。但是排序后的索引变成了排名而不是旧的索引。顺便说一句，我使用top 来获取RDD 中的前n 个有序元素。似乎它的工作原理类似于takeOrdered。