如何解释 RDD.treeAggregate答案

【问题标题】：how to interpret RDD.treeAggregate如何解释 RDD.treeAggregate
【发布时间】：2015-07-03 20:00:46
【问题描述】：

我在 Apache Spark 代码源中遇到了this line

val (gradientSum, lossSum, miniBatchSize) = data
    .sample(false, miniBatchFraction, 42 + i)
    .treeAggregate((BDV.zeros[Double](n), 0.0, 0L))(
      seqOp = (c, v) => {
        // c: (grad, loss, count), v: (label, features)
        val l = gradient.compute(v._2, v._1, bcWeights.value, Vectors.fromBreeze(c._1))
        (c._1, c._2 + l, c._3 + 1)
      },
      combOp = (c1, c2) => {
        // c: (grad, loss, count)
        (c1._1 += c2._1, c1._2 + c2._2, c1._3 + c2._3)
      }
    )

我在阅读这篇文章时遇到了很多问题：

首先，我在网上找不到任何可以准确解释treeAggregate 工作原理的东西，参数的含义是什么。
其次，这里.treeAggregate 方法名后面好像有两个()()。那意味着什么？那是我不明白的一些特殊的scala语法吗？
最后，我看到 seqOp 和 comboOp 都返回一个 3 元素元组，它与预期的左侧变量匹配，但实际上返回的是哪一个？

这句话一定很高级。我无法开始解读这个。

【问题讨论】：

标签： scala apache-spark rdd distributed-computing

【解决方案1】：

treeAggregate 是aggregate 的专门实现，它迭代地将组合函数应用于分区子集。这样做是为了防止像经典的aggregate 那样将所有部分结果返回给驱动程序，在这种情况下会发生单程减少。

出于所有实际目的，treeAggregate 遵循与aggregate 在此答案中解释的相同原则：Explain the aggregate functionality in Python，但它需要一个额外的参数来指示部分聚合级别的深度。

让我试着具体解释一下这里发生了什么：

对于聚合，我们需要一个零、一个组合函数和一个归约函数。 aggregate 使用 currying 独立于 combine 和 reduce 函数指定零值。

然后我们可以像这样剖析上面的函数。希望这有助于理解：

val Zero: (BDV, Double, Long) = (BDV.zeros[Double](n), 0.0, 0L)
val combinerFunction: ((BDV, Double, Long), (??, ??)) => (BDV, Double, Long)  =  (c, v) => {
        // c: (grad, loss, count), v: (label, features)
        val l = gradient.compute(v._2, v._1, bcWeights.value, Vectors.fromBreeze(c._1))
        (c._1, c._2 + l, c._3 + 1)
val reducerFunction: ((BDV, Double, Long),(BDV, Double, Long)) => (BDV, Double, Long) = (c1, c2) => {
        // c: (grad, loss, count)
        (c1._1 += c2._1, c1._2 + c2._2, c1._3 + c2._3)
      }

然后我们可以以更易于理解的形式重写对treeAggregate的调用：

val (gradientSum, lossSum, miniBatchSize) = treeAggregate(Zero)(combinerFunction, reducerFunction)

此表单会将生成的元组“提取”到命名值gradientSum, lossSum, miniBatchSize 以供进一步使用。

请注意，treeAggregate 带有一个附加参数 depth，该参数声明为默认值 depth = 2，因此，由于此特定调用中未提供该参数，因此它将采用该默认值。

【讨论】：