让我们将其分解为离散的方法和类型,通常这样可以为新开发人员展示其复杂性:
pairs.reduceByKey((a, b) => a + b)
pairs.reduceByKey((a: Int, b: Int) => a + b)
重命名变量可以使其更加明确
pairs.reduceByKey((accumulatedValue: Int, currentValue: Int) => accumulatedValue + currentValue)
因此,我们现在可以看到,我们只是取给定键的累积值,并将其与该键的下一个值相加。现在,让我们进一步分解它,以便我们可以理解其中的关键部分。因此,让我们更多地像这样来可视化这种方法:
pairs.reduce((accumulatedValue: List[(String, Int)], currentValue: (String, Int)) => {
val accumAsMap = accumulatedValue.toMap
accumAsMap.get(currentValue._1) match {
case Some(value : Int) => (accumAsMap + (currentValue._1 -> (value + currentValue._2))).toList
case None => currentValue :: accumulatedValue
}
})
因此,您可以看到,reduceByKey消除了查找键和跟踪键的模板,使您无需担心管理该部分。
如果您想更深入、更准确
尽管如此,这只是发生的简化版本,因为这里有一些优化。这个操作是可结合的,所以Spark引擎会首先在本地执行这些缩减操作(通常称为地图端缩减),然后再次在驱动程序中执行。这节省了网络流量;而不是发送所有数据并执行操作,它可以将其缩小到最小,并将该缩减发送到线路上。