Spark: Получить Top N по ключевым

Question

Say У меня есть PairRDD как таковой (Очевидно, гораздо больше данных, в реальной жизни, предположим, что миллионы записей):

val scores = sc.parallelize(Array(
     ("a", 1), 
     ("a", 2), 
     ("a", 3), 
     ("b", 3), 
     ("b", 1), 
     ("a", 4), 
     ("b", 4), 
     ("b", 2) 
)) 

Что является наиболее эффективным способом для создания RDD с верхние 2 счета за ключ?

val top2ByKey = ... 
res3: Array[(String, Int)] = Array((a,4), (a,3), (b,4), (b,3)) 

Answer 1

Я думаю, что это должно быть весьма эффективным:

Отредактировано согласно комментариям OP:

scores.mapValues(p => (p, p)).reduceByKey((u, v) => { 
    val values = List(u._1, u._2, v._1, v._2).sorted(Ordering[Int].reverse).distinct 
    if (values.size > 1) (values(0), values(1)) 
    else (values(0), values(0)) 
}).collect().foreach(println) 

Answer 2

scores.reduceByKey(_ + _).map(x => x._2 -> x._1).sortByKey(false).map(x => x._2 -> x._1).take(2).foreach(println) 

Answer 3

Начиная с версии 1.4, есть встроенный способ сделать это с использованием MLLib: https://github.com/apache/spark/blob/master/mllib/src/main/scala/org/apache/spark/mllib/rdd/MLPairRDDFunctions.scala

import org.apache.spark.mllib.rdd.MLPairRDDFunctions.fromPairRDD 
scores.topByKey(2) 

Answer 4

Немного изменены ваши входные данные.

val scores = sc.parallelize(Array(
     ("a", 1), 
     ("a", 2), 
     ("a", 3), 
     ("b", 3), 
     ("b", 1), 
     ("a", 4), 
     ("b", 4), 
     ("b", 2), 
     ("a", 6), 
     ("b", 8) 
    )) 

Я объясню, как это сделать шаг за шагом:

1.Group ключом для создания массива

scores.groupByKey().foreach(println) 

Результат:

(b,CompactBuffer(3, 1, 4, 2, 8)) 
(a,CompactBuffer(1, 2, 3, 4, 6)) 

Как вы видите, каждый само значение представляет собой массив чисел. CompactBuffer - это просто оптимизированный массив.

2.Для каждый ключ, реверс сортировки списка чисел, значение содержит

scores.groupByKey().map({ case (k, numbers) => k -> numbers.toList.sorted(Ordering[Int].reverse)}).foreach(println) 

Результат:

(b,List(8, 4, 3, 2, 1)) 
(a,List(6, 4, 3, 2, 1)) 

3.Keep только первые 2 элемента из 2-й стадии, они будут сверху 2 баллов в списке

scores.groupByKey().map({ case (k, numbers) => k -> numbers.toList.sorted(Ordering[Int].reverse).take(2)}).foreach(println) 

Результат:

(a,List(6, 4)) 
(b,List(8, 4)) 

4.Flat карта для создания новых спаренных RDD для каждого ключа и наивысший балл

scores.groupByKey().map({ case (k, numbers) => k -> numbers.toList.sorted(Ordering[Int].reverse).take(2)}).flatMap({case (k, numbers) => numbers.map(k -> _)}).foreach(println) 

Результат:

(b,8) 
(b,4) 
(a,6) 
(a,4) 

5.Optional шаг - сортировка по ключу, если вы хотите

scores.groupByKey().map({ case (k, numbers) => k -> numbers.toList.sorted(Ordering[Int].reverse).take(2)}).flatMap({case (k, numbers) => numbers.map(k -> _)}).sortByKey(false).foreach(println) 

Результаты:

(a,6) 
(a,4) 
(b,8) 
(b,4) 

Надеюсь, это объяснение помогло понять логику.