Понимание функции JavaPairRDD.reduceByKey

Question

Я наткнулся на следующий фрагмент кода Apache Spark:

JavaRDD<String> lines = new JavaSparkContext(sparkSession.sparkContext()).textFile("src\\main\\resources\\data.txt");
JavaPairRDD<String, Integer> pairs = lines.mapToPair(s -> new Tuple2(s, 1));
System.out.println(pairs.collect());
JavaPairRDD<String, Integer> counts = pairs.reduceByKey((a, b) -> a + b);
System.out.println("Reduced data: " + counts.collect());

Мой data.txt выглядит следующим образом:

Mahesh
Mahesh
Ganesh
Ashok
Abnave
Ganesh
Mahesh

Вывод:

[(Mahesh,1), (Mahesh,1), (Ganesh,1), (Ashok,1), (Abnave,1), (Ganesh,1), (Mahesh,1)]
Reduced data: [(Ganesh,2), (Abnave,1), (Mahesh,3), (Ashok,1)]

Хотя я понимаю, как получается первая строка вывода, я не понимаю, как получается вторая строка, то есть, как JavaPairRDD<String, Integer> counts образовано reduceByKey.

Я обнаружил, что подпись из reduceByKey() выглядит следующим образом:

public JavaPairRDD<K,V> reduceByKey(Function2<V,V,V> func)

[подпись] (http://spark.apache.org/docs/1.2.0/api/java/org/apache/spark/api/java/function/Function2.html#call(T1, T2)) Function2.call() выглядит следующим образом:

R call(T1 v1, T2 v2) throws Exception

Объяснение reduceByKey() гласит следующее:

Объедините значения для каждого ключа, используя ассоциативную функцию сокращения.Это также выполнит слияние локально для каждого преобразователя перед отправкой результатов в редуктор, аналогично «объединителю» в MapReduce.Вывод будет разделен на хеш с существующим уровнем разделения / параллелизма.

Теперь это объяснение звучит несколько смущающим для меня.Может быть, есть что-то еще в функциональности reduceByKey().Глядя на ввод и вывод в reduceByKey() и Function2.call(), я чувствую, что reducebyKey() посылает значения тех же клавиш в call() парами.Но это просто не звучит ясно.Кто-нибудь может объяснить, что именно, как reduceByKey() и Function2.call() работают вместе?

Answer 1

Я думаю, что вы, наверное, понимаете groupByKey? groupByKey группирует все значения для определенного ключа в список (или итеративный), чтобы вы могли что-то делать с этим - например, суммировать (или считать) значения. По сути, sum делает сокращение списка многих значений в одно значение. Это достигается путем итеративного добавления двух значений, чтобы получить одно значение, и это то, что Function2 необходимо сделать, когда вы пишете свое собственное. Он должен принимать два значения и возвращать одно значение.

ReduceByKey делает то же самое, что и groupByKey, НО он выполняет то, что называется «уменьшение на стороне карты» , прежде чем перетасовывать данные. Поскольку Spark распределяет данные по многим различным компьютерам для параллельной обработки, нет гарантии, что данные с одним и тем же ключом будут размещены на одном компьютере. Spark, таким образом, должен перетасовывать данные, и чем больше данных нужно перетасовать, тем дольше будут выполняться наши вычисления, поэтому рекомендуется перетасовывать как можно меньше данных.

При уменьшении на стороне карты , Spark сначала суммирует все значения для данного ключа локально на исполнителях, прежде чем отправит (перемешивает) результат вокруг для получения окончательной суммы быть вычисленным. Это означает, что гораздо меньшее количество данных - одно значение вместо списка значений - необходимо отправлять между различными машинами в кластере, и по этой причине reduceByKey чаще всего предпочтительнее, чем groupByKey.

Для более подробного описания могу порекомендовать эту статью :)

Answer 2

Как следует из названия, reduceByKey() уменьшает данные в зависимости от лямбда-функции, которую вы передаете ей.

В вашем примере эта функция представляет собой простой сумматор: для a и b возвращаем a + b. Лучший способ понять, как формируется результат, - представить, что происходит внутри. Часть ByKey() группирует ваши записи на основе их ключевых значений. В вашем примере у вас будет 4 разных набора пар:

Набор 1: ((Mahesh, 1), (Mahesh, 1), (Mahesh, 1))

Набор 2: ((Ganesh, 1), (Ganesh, 1))

Набор 3: ((Ashok, 1))

Набор 4: ((Abnave, 1))

Теперь часть reduce попытается уменьшить предыдущие 4 набора, используя лямбда-функцию (сумматор):

Для набора 1: (Махеш, 1 + 1 + 1) -> (Махеш, 3)

Для набора 2: (Ганеша, 1 + 1) -> (Ганеша, 2)

Для набора 3: (Ашок, 1) -> (Ашок, 1) (добавить нечего)

Для набора 4: (Abnave, 1) -> (Abnave, 1) (добавить нечего)

Сигнатуры функций иногда могут сбивать с толку, поскольку они имеют тенденцию быть более общими.