apache-spark - 自定义聚合器中的 Spark 累加器

Question

我有一个自定义聚合器，在 reduce 方法中我想使用累加器来做一些统计。

我应该如何将累加器传递给聚合器？

我必须将累加器作为构造函数参数传递还是必须使用AccumulatorContext.get(0)？

Answer 1

在聚合器之外（以及将在执行程序节点上运行的任何其他代码之外）创建累加器，并在聚合器中使用它。变量本身可以像任何其他普通变量一样传递。

val ds = sc.parallelize(1 to 5).toDS()
val acc = sc.longAccumulator
val mySumAgg = new Aggregator[Int, Int, Int] {
  def reduce(b: Int, a: Int): Int = {
    acc.add(1)
    a + b
  }      
  [...]
}.toColumn

ds.groupByKey( i => i)
  .agg(mySumAgg)
  .show()

print("Merge has been called " + acc.value + " times")

如果您为聚合器创建了一个单独的类，则可以通过构造函数传递累加器，也可以使用 setter。

正如文档所说，您不应该使用AccumulatorContext：

Spark 本身用于跟踪累加器的内部类。