Spark 对 RDD 进行排序并加入他们的行列

Question

我有一个 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

Answer 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))

此致，

Spark 对 RDD 进行排序并加入他们的行列

Spark sort RDD and join their rank

scala

apache-spark

rdd