Спарк Scala: GroupByKey и сортировки

Question

У меня есть RDD со следующей структурой:

val rdd = RDD[ (category: String, product: String, score: Double) ] 

Моя цель состоит в том, чтобы group данных, основанных на категории, а затем для каждой категории sort w.r.t. оценка Tuple 2 (product, score). На данный момент мой код:

val result = rdd.groupByKey.mapValues(v => v.toList.sortBy(-_._2)) 

Это оказывается очень дорогостоящей операцией для данных, которые у меня есть. Я ищу улучшить производительность, используя альтернативный подход.

Answer 1

Трудно ответить, не зная, набор данных, но documentation имеет некоторые подсказки повторно: groupByKey производительность:

Примечание: Эта операция может быть очень дорогим. Если вы группируете в заказ для выполнения агрегирования (например, суммы или среднего) по каждому ключу , используя PairRDDFunctions.aggregateByKey или PairRDDFunctions.reduceByKey обеспечит гораздо лучшую производительность.

Так что это зависит от того, что вы собираетесь делать с отсортированными списками. Если вам нужен весь список, то его может быть сложно улучшить на groupByKey. Если вы выполняете какую-то агрегацию, то альтернативные операции выше (aggregateByKey, reduceByKey) могут быть лучше.

Перед сортировкой в ​​зависимости от размера ваших списков может быть может использовать более эффективную альтернативную коллекцию (например, изменяемый массив).

Редактировать: Если у вас относительно небольшое количество категорий, вы можете попробовать несколько раз фильтровать исходную RDD и сортировать каждый отфильтрованный RDD. Хотя подобный объем работы выполнен в целом, он может использовать меньше памяти в любой момент.

Редактировать 2: Если недостаток памяти является проблемой, вы можете представить свои категории и продукты как целые идентификаторы, а не строки, и только искать имена позже. Таким образом, ваш основной RDD может быть намного меньше.

Answer 2

Является ли ваше RDD справедливым распределенным по категориям? У вас могут быть проблемы в зависимости от вашего фактора перекоса. Попробуйте что-то вроде этого, если вы не слишком много ключевых значений:

val rdd: RDD[(String, String, Double)] = sc.parallelize(Seq(("someCategory","a",1.0),("someCategory","b",3.0),("someCategory2","c",4.0))) 

rdd.keyBy(_._1).countByKey().foreach(println)