scala - 如何将 Spark 的累加器传递给函数？

Question

我想做这样的事情。

val ac = sc.accumulator(0)
....
a = a.map(x => someFunction(x, the_accumulator_object))
....

the_accumulator_ojbect的位置应该是什么在上面的代码中？会写 ac有就好了吗？

另外，在函数中

def someFunction(x: TypeOfX, a: TypeOfAccumulator) : ReturnType =
{
    .....
}

TypeOfAccumulator的位置应该是什么在上面的函数中？

Answer 1

有关 Spark 蓄电池的更多信息，请访问 here

根据有关创建累加器的 scala-docs :

/** * Create an [[org.apache.spark.Accumulator]] variable of a given type, with a name for display * in the Spark UI. Tasks can "add" values to the accumulator using the += method. Only the * driver can access the accumulator's value. */

默认累加器类型为 int .您可以设置自己的类型，但需要正确实现 +=将值添加到您自己的累加器类型的方法:

val ac = sc.accumulator[MyOwnType](MyOwnTypeObject, "my own type object accumulator")

您的主要代码片段将如下所示:

val ac = sc.accumulator(0, "some accumulator")
....
a = a.map(x => someFunction(x, ac))
....
System.out.println("My accumulator value is: " + ac.value)

凡 someFunction方法植入将类似于:

def someFunction(x: TypeOfX, ac: Accumulator[Int]) : ReturnType =
{
    ...
    ac += 1
    ...
}