何时使用 countByValue 以及何时使用 map().reduceByKey()答案

【问题标题】：When to use countByValue and when to use map().reduceByKey()何时使用 countByValue 以及何时使用 map().reduceByKey()
【发布时间】：2019-03-25 17:01:24
【问题描述】：

我是 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()

这也很好用。

现在，我的问题是何时使用哪些方法？哪一个比另一个更受欢迎？

【问题讨论】：

您是否认为答案符合您对知识的要求？
知道是否接受答案吗？
在声明中： val wordCounts = words.map(x => (x, 1)).reduceByKey((x, y) => x + y) 有人可以解释为什么 x + y 吗？

标签： scala apache-spark rdd word-count

【解决方案1】：

countByValue() 是一个 RDD 操作，它返回每个唯一的计数此 RDD 中的值作为 (value, count) 对的字典。
reduceByKey() 是一种 RDD 转换，它以对的格式返回 RDD

【讨论】：

【解决方案2】：

除了以上所有答案，我还发现了以下内容：

CountByValue 返回一个不能以分布式方式使用的地图。
ReduceByKey 返回一个 rdd，可以进一步以分布式方式使用。

【讨论】：

【解决方案3】：

至少在 PySpark 中，它们是不同的东西。

countByKey 是用reduce 实现的，这意味着驱动程序将收集分区的部分结果并自己进行合并。如果你的结果很大，那么驱动就得合并大量的大字典，这会让驱动发疯。

reduceByKey 将 key 打乱到不同的 executor 中，并在每个 worker 中做 reduce，所以如果数据很大的话更有利。

总之，当你的数据很大时，使用map、reduceByKey和collect会让你的司机更快乐。如果您的数据很小，countByKey 将引入更少的网络流量（少一个阶段）。

【讨论】：

【解决方案4】：

这里的例子 - 不是文字，而是数字：

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)

这是关键的区别。

如果您想要一个开箱即用的地图，那么这就是您要走的路。

另外，关键是reduce是隐含的，不能被影响，也不需要像reduceByKey那样提供。

当数据量很大时，reduceByKey 具有优先权。 Map 被完整地加载到 Driver 内存中。

【讨论】：