需要返回R1？或者有什么价值？

Question

“可累积”的星火源代码中的addInPlace方法将来自不同分区的相同可累积值合并：

/**
 * Merge two accumulated values together. Is allowed to modify and return the first value
 * for efficiency (to avoid allocating objects).
 *
 * @param r1 one set of accumulated data
 * @param r2 another set of accumulated data
 * @return both data sets merged together
 */
def addInPlace(r1: R, r2: R): R

当我在AccumulableParam的实现中定义AccumulableParam时，我假设我可以返回我想要的任何值。我假设r1传入的任何指针都会指向我返回的任何指针。

我的老板认为r1的传入是返回语句中唯一允许的内容。这听起来像安·兰德斯，谁是对的？

有一种情况是，我只想丢弃r1并将其替换为r2中的对象，这将是这个合并累加器的新值。

我可以返回r2吗?或者我必须像我的老板(经验多得多的Java编程)所想的那样，对r1做一个深入的复制？要清楚，虽然Spark当然是用Scala编写的，但我正在编写一个用Java实现AccumulableParam的类。

Answer 1

根据经验，在执行类折叠操作时，永远不会修改第二个参数。我们可以用一个简单的例子来说明为什么。让我们假设我们有这样的简单累加器：

import org.apache.spark.AccumulatorParam
import scala.collection.mutable.{Map => MMap}

type ACC = MMap[String, Int]

object DummyAccumulatorParam extends AccumulatorParam[ACC] {
  def zero(initialValue: ACC): ACC = {
    initialValue
  }

  def addInPlace(acc: ACC, v: ACC): ACC = {
    v("x") = acc.getOrElse("x", 0) +  v.getOrElse("x", 0)
    v
  }
}

它特别有用，但并不重要。重点是它修改了第二个论点。让我们看看它是否有效：

val rdd = sc.parallelize(Seq(MMap("x" -> 1), MMap("x" -> 1), MMap("x" -> 1)), 1)

val accum1 = sc.accumulator(MMap("x" -> 0))(DummyAccumulatorParam)
rdd.foreach(x => accum1 += x)

accum1.value
// scala.collection.mutable.Map[String,Int] = Map(x -> 3)

到目前一切尚好。我们甚至可以创建另一个，它可以像预期的那样工作：

val accum2 = sc.accumulator(MMap("x" -> 0))(DummyAccumulatorParam)
rdd.foreach(x => accum2 += x)

accum2.value
// scala.collection.mutable.Map[String,Int] = Map(x -> 3)

现在让我们缓存数据：

rdd.cache

重复这一过程：

val accum3 = sc.accumulator(MMap("x" -> 0))(DummyAccumulatorParam)
rdd.foreach(x => accum3 += x)

val accum4 = sc.accumulator(MMap("x" -> 0))(DummyAccumulatorParam)
rdd.foreach(x => accum4 += x)

并检查累加器值：

accum4.value
// scala.collection.mutable.Map[String,Int] = Map(x -> 6)

和RDD内容：

rdd.collect
// Array[scala.collection.mutable.Map[String,Int]] = 
//  Array(Map(x -> 1), Map(x -> 3), Map(x -> 6))

因此，如您所见，返回或修改第二个参数是不安全的。它也适用于其他类似的操作，如fold或aggregate。