Spark 对 RDD 进行排序并加入他们的行列
Spark sort RDD and join their rank
我有一个 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
如果有任何机会,您只想将 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))
此致,
我有一个 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
如果有任何机会,您只想将 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))
此致,