Spark Pair RDD中按值排序

Question

我有一个如下所示的 Spark 对偶 RDD（键，计数）

Array[(String, Int)] = Array((a,1), (b,2), (c,1), (d,3))

如何使用Spark Scala API获取按值排序的新对RDD？

所需结果：Array((d,3), (b,2), (a,1), (c,1))

- Vijay Innamuri

---

值中可能会有重复吗？如果没有：交换键和值，按键排序，然后再交换回来。 - Gábor Bakos

是的，值是重复的。 - Vijay Innamuri

Answer 1

这应该可以工作：

//Assuming the pair's second type has an Ordering, which is the case for Int
rdd.sortBy(_._2) // same as rdd.sortBy(pair => pair._2)

（虽然当出现平局时，您可能也希望考虑密钥。）

- Gábor Bakos

---

嗨@GáborBakos，我创建了一个像这样的映射 - val b = a.map(x => (x(0), x) ) 这里的b是类型为org.apache.spark.rdd.RDD[(Any, org.apache.spark.sql.Row)]。当我尝试b.sortBy(_._2)时它不起作用。是否有关于如何使用_的讨论？ - user 923227

嗨@SumitKumarGhosh，这似乎是一个不同的问题。我认为您需要为org.apache.spark.sql.Row类型定义一个“Ordering”，因为它不是“Ordered”。如果您创建一个新问题，请同时添加错误消息。下划线“”只是以下语法糖：“aVariable => aVariable”，因此“._2”也可以像这样编写：“pair => pair._2”。 - Gábor Bakos

请参考以下链接，其中讨论了如何在Spark PairRDD中按值排序，其中值来自Spark SQL DataFrame。 - user 923227

针对那些搜寻 PySpark 解决方案的读者，请注意：rdd.sortBy(lambda pair:pair[1]) 的意思是按照元组中第二个元素进行排序。 - Ébe Isaac

要降序排列，使用：rdd.sortBy(_._2, false)。 参考链接：https://dev59.com/ElgQ5IYBdhLWcg3w-48d。 - Omkar Neogi

Answer 2

按键和值按升序和降序排序

val textfile = sc.textFile("file:///home/hdfs/input.txt")
val words = textfile.flatMap(line => line.split(" "))
//Sort by value in descending order. For ascending order remove 'false' argument from sortBy
words.map( word => (word,1)).reduceByKey((a,b) => a+b).sortBy(_._2,false)
//for ascending order by value
words.map( word => (word,1)).reduceByKey((a,b) => a+b).sortBy(_._2)

//Sort by key in ascending order
words.map( word => (word,1)).reduceByKey((a,b) => a+b).sortByKey
//Sort by key in descending order
words.map( word => (word,1)).reduceByKey((a,b) => a+b).sortByKey(false)

将键和值交换后，可以通过应用sortByKey的方式以另一种方式完成此操作。

//Sort By value by swapping key and value and then using sortByKey
val sortbyvalue = words.map( word => (word,1)).reduceByKey((a,b) => a+b)
val descendingSortByvalue = sortbyvalue.map(x => (x._2,x._1)).sortByKey(false)
descendingSortByvalue.toDF.show
descendingSortByvalue.foreach {n => {
val word=  n._1
val count = n._2
println(s"$word:$count")}}

- Nagaraj Vittal

---

感谢Vijay Innamuri。 - Nagaraj Vittal

在后一种情况下，应该像这样进行吗？ val descendingSortByvalue = sortbyvalue.map(x => (x._2,x._1)).sortByKey(false) - Pawel Kranzberg