Когда использовать countByValue, а когда использовать map (). ReduceByKey () - PullRequest
Новая
       32

Когда использовать countByValue, а когда использовать map (). ReduceByKey ()

0 голосов
/ 21 октября 2018

Я новичок в Spark и scala и работаю над простым примером WordCount.

Так что для этого я использую countByValue следующим образом: 

val words = lines.flatMap(x => x.split("\\W+")).map(x => x.toLowerCase())
val wordCount = words.countByValue();

, который отлично работает.

И то же самое может быть достигнуто следующим образом: 

val words = lines.flatMap(x => x.split("\\W+")).map(x => x.toLowerCase())
val wordCounts = words.map(x => (x, 1)).reduceByKey((x, y) => x + y)
val sortedWords = wordCounts.map(x => (x._2, x._1)).sortByKey()

, который также отлично работает.

Теперь мой вопрос: когда использовать какие методы?Какой из них предпочтительнее другого?

Ответы [ 3 ]

0 голосов
/ 16 февраля 2019

По крайней мере, в PySpark это разные вещи.

countByKey реализовано с reduce, что означает, что драйвер будет собирать частичные результаты разделов и выполняет слияние.Если ваш результат велик, то драйверу придется объединить большое количество больших словарей, что сведет его с ума.

reduceByKey перетасовывает ключи для разных исполнителей и выполняет сокращение для каждого работника,так что лучше, если данные большие.

В заключение, если ваши данные велики, используйте map, reduceByKey и collect, что сделает ваш драйвер намного счастливее.Если ваши данные невелики, countByKey представит меньше сетевого трафика (на одну ступень меньше).

0 голосов
/ 15 марта 2019

В дополнение ко всем ответам, приведенным выше, я нашел следующее:

  1. CountByValue возвращает карту, которую нельзя использовать распределенным способом.

  2. ReduceByKey возвращает rdd, который в дальнейшем может использоваться распределенным образом.

0 голосов
/ 21 октября 2018

Пример здесь - не слова, а цифры: 

val n = sc.parallelize(List(1,2,3,4,5,6,7,8,2,4,2,1,1,1,1,1))
val n2 = n.countByValue

возвращает локальную карту: 

n: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[4] at parallelize at command-3737881976236428:1
n2: scala.collection.Map[Int,Long] = Map(5 -> 1, 1 -> 6, 6 -> 1, 2 -> 3, 7 -> 1, 3 -> 1, 8 -> 1, 4 -> 2)

В этом ключевое отличие.

Если вы хотите получить карту из коробки, то это путь.

Кроме того, дело в том, что уменьшение подразумевается и не может быть затронуто, и его не нужно предоставлять, как в reduByKey.

ReduByKey имеет преимущество при больших размерах данных.Карта полностью загружена в память драйвера.

...