scala – 如何解释RDD.treeAggregate

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

导读：我在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) va

我在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,count)
        (c1._1 += c2._1,c1._2 + c2._2,c1._3 + c2._3)
      }
    )

我读这个有多个麻烦：

>首先,我在网上找不到任何解释treeAggregate如何工作的内容,params的含义是什么.
>其次,这里.treeAggregate似乎在方法名后面有两个()().这意味着什么？这是一些我不理解的特殊scala语法.
>最后,我看到seqOp和comboOp都返回一个3元素元组,它与预期的左侧变量匹配,但实际返回哪一个？

这个陈述必须非常先进.我无法开始破译这一点.

解决方法

treeAggregate是聚合的专用实现,它将组合函数迭代地应用于分区的子集.这样做是为了防止将所有部分结果返回给驱动程序,其中单个传递减少将像经典聚合那样发生.

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

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

对于聚合,我们需要零,组合器函数和reduce函数.
aggregate使用currying独立于combine和reduce函数指定零值.

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

val Zero: (BDV,Double,Long) = (BDV.zeros[Double](n),0L)
val combinerFunction: ((BDV,Long),(??,??)) => (BDV,Long)  =  (c,c._3 + 1)
val reducerFunction: ((BDV,(BDV,Long)) => (BDV,Long) = (c1,c1._3 + c2._3)
      }

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

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

此表单将“生成”结果元组到命名值gradientSum,miniBatchSize以供进一步使用.

请注意,treeAggregate接受一个额外的参数depth,该参数使用默认值depth = 2声明,因此,在此特定调用中未提供该参数,它将采用该默认值.

（编辑：李大同）

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