在Spark RDD中获取每个键的最大值

Question

在Spark RDD中，返回每个唯一键关联的最大行（值）的最佳方法是什么？

我正在使用Python，并尝试使用Math max、映射和按键减少聚合。有没有有效的方法来做到这一点？可能是一个UDF？

我有一个RDD格式：

[(v, 3),
 (v, 1),
 (v, 1),
 (w, 7),
 (w, 1),
 (x, 3),
 (y, 1),
 (y, 1),
 (y, 2),
 (y, 3)]

我需要您返回：

[(v, 3),
 (w, 7),
 (x, 3),
 (y, 3)]

连接可以返回第一个值或随机值。

- captainKirk104

Answer 1

实际上你有一个PairRDD，使用reduceByKey是最好的方法之一：

(Scala)

val grouped = rdd.reduceByKey(math.max(_, _))

(Python)

grouped = rdd.reduceByKey(max)

（Java 7）

JavaPairRDD<String, Integer> grouped = new JavaPairRDD(rdd).reduceByKey(
    new Function2<Integer, Integer, Integer>() {
        public Integer call(Integer v1, Integer v2) {
            return Math.max(v1, v2);
    }
});

（Java 8）

JavaPairRDD<String, Integer> grouped = new JavaPairRDD(rdd).reduceByKey(
    (v1, v2) -> Math.max(v1, v2)
);

reduceByKey的API文档：

• Scala
• Python
• Java

- Daniel de Paula

---

你能提供一种在Java中实现的方法吗？我正在使用Java，并且正在寻找完全相同的东西。 - tsar2512

使用Java 8，可以这样实现：new JavaPairRDD(rdd).reduceByKey((v1, v2) -> Math.max(v1, v2)); - Daniel de Paula

谢谢你的回应，不幸的是，我正在使用Java 7 - 它不允许Lambda函数。通常必须编写匿名函数。你能告诉我在Java 7中怎么解决吗？我猜一个简单的比较函数应该可以解决！ - tsar2512

另外，我们得到的是属于每个键的值的最大值。这正确吗？ - tsar2512

@tsar2512，是的，生成的RDD将包含每个键的单个条目，其中包含一对（key，maxValue）。我已更新答案，提供了Java 7和Java 8的版本，但我尚未测试它们，请告诉我是否有效。 - Daniel de Paula

@DanieldePaula 如果初始RDD具有元组，例如[(v,(125,12)), (v, (130,15)), (w,(200,30)), (w,(250,40))]，而不仅仅是一个值，并且您需要查找第一个元素的最大值结果，例如[(v, (130,15)),(w,(250,40))]，该怎么办？ - user2829319