Как получить минимальное значение для каждого отдельного ключа, используя ReduceByKey () в Scala

Question

У меня есть плоская карта, которая возвращает последовательность Seq((20,6),(22,6),(23,6),(24,6),(20,1),(22,1)), теперь мне нужно использовать reduceByKey() в последовательности, которую я получил от плоской карты, чтобы найти минимальное значение для каждого ключа.

Iпопытался использовать .reduceByKey(a,min(b)) и .reduceByKey((a, b) => if (a._1 < b._1) a else b), но ни один из них не работает.

Это мой код

for(i<- 1 to 5){

var graph=graph.flatMap{ in => in match{ case (x, y, zs) => (x, y) :: zs.map(z => (z, y))}
.reduceByKey((a, b) => if (a._1 < b._1) a else b)

}

Для каждого отдельного ключа, который генерирует плоская карта, мне нужно получить минимальное значение дляэтот ключ.Например: плоская карта генерирует Seq ((20,6), (22,6), (23,6), (24,6), (20,1), (22,1)), который должен генерировать resultByKey () (20,1), (22,1), (23,6), (24,6)

Answer 1

Вот подпись reduceByKey:

def reduceByKey(func: (V, V) ⇒ V): RDD[(K, V)]

По сути, учитывая СДР пар ключ / значение, вам необходимо предоставить функцию, которая сокращает два значения (а не всю пару) доодин.Поэтому вы можете использовать его следующим образом:

val rdd = sc.parallelize(Seq((20,6),(22,6),(23,6),(24,6),(20,1),(22,1)))
val result = rdd.reduceByKey((a, b) => if (a < b) a else b)
result.collect
// Array[(Int, Int)] = Array((24,6), (20,1), (22,1), (23,6))