scala – reduceByKey：它在内部如何工作？

发布时间：2020-12-16 09:29:47 所属栏目：安全来源：网络整理

导读：我是Spark和Scala的新手。我对reduceByKey函数在Spark中的工作方式感到困惑。假设我们有以下代码： val lines = sc.textFile("data.txt")val pairs = lines.map(s = (s,1))val counts = pairs.reduceByKey((a,b) = a + b) map函数是明确的：s是键，它指向dat

我是Spark和Scala的新手。我对reduceByKey函数在Spark中的工作方式感到困惑。假设我们有以下代码：

val lines = sc.textFile("data.txt")
val pairs = lines.map(s => (s,1))
val counts = pairs.reduceByKey((a,b) => a + b)

map函数是明确的：s是键，它指向data.txt中的行，1是值。

但是，我没有得到reduceByKey如何在内部工作？ “a”指向钥匙吗？或者，“a”指向“s”吗？那么什么代表一个b？它们是如何填满的？

解决方法

让我们分解为离散的方法和类型。这通常暴露了新开发者的错综复杂：

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)) => {
  //Turn the accumulated value into a true key->value mapping
  val accumAsMap = accumulatedValue.toMap   
  //Try to get the key's current value if we've already encountered it
  accumAsMap.get(currentValue._1) match { 
    //If we have encountered it,then add the new value to the existing value and overwrite the old
    case Some(value : Int) => (accumAsMap + (currentValue._1 -> (value + currentValue._2))).toList
    //If we have NOT encountered it,then simply add it to the list
    case None => currentValue :: accumulatedValue 
  }
})

因此，您可以看到reduceByKey采用寻找密钥并跟踪它的样板，因此您不必担心管理该部分。

更深入，更真实，如果你想

所有这一切，这是一个简化的版本，因为这里有一些优化。此操作是关联的，因此火花引擎将首先在本地执行这些减少(通常称为地图侧减少)，然后再次在驾驶员处执行。这节省了网络流量;而不是发送所有数据并执行操作，它可以尽可能小地减少它，然后通过线路发送减少量。

（编辑：李大同）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!