为什么工作节点看不到另一个工作节点上累加器的更新？

Question

我LongAccumulator在地图操作中使用a 作为共享计数器。但是似乎我没有正确使用它，因为工作节点上计数器的状态没有更新。这是我的柜台类的样子：

public class Counter implements Serializable {

   private LongAccumulator counter;

   public Long increment() {
      log.info("Incrementing counter with id: " + counter.id() + " on thread: " + Thread.currentThread().getName());
      counter.add(1);
      Long value = counter.value();
      log.info("Counter's value with id: " + counter.id() + " is: " + value + " on thread: " + Thread.currentThread().getName());
      return value;
   }

   public Counter(JavaSparkContext javaSparkContext) {
      counter = javaSparkContext.sc().longAccumulator();
   }
}

据我了解的文档，当应用程序在多个工作程序节点中运行时，这应该可以正常工作：

累加器是仅通过关联和交换操作“累加”的变量，因此可以有效地并行支持。它们可用于实现计数器（如在MapReduce中）或总和。Spark本身支持数字类型的累加器，程序员可以添加对新类型的支持。

但是这是当计数器在2个不同的工作线程上递增并且看起来状态未在节点之间共享时的结果：

INFO计数器：线程上ID为866的递增计数器：Executor任务启动worker-6
INFO计数器：线程ID为866的计数器值为：1线程：Executor任务启动worker-6
INFO计数器：ID为866的递增计数器，线程：执行程序任务启动worker-0
INFO计数器：ID：866的计数器的值是：1在线程上：执行程序任务启动worker-0

我是否理解累加器概念错误，或者必须使用任何设置来启动任务？

Answer 1

它不应该工作：

然后，可以使用add方法将在集群上运行的任务添加到集群中。但是，他们无法读取其值。只有驱动程序可以使用value方法读取累加器的值。

每个任务都有其自己的累加器，该累加器在本地更新，并在完成任务并报告结果后与驱动程序上的“共享”副本合并。

从任务中使用旧AccumulatorAPI时（AccumulatorV2实际上是包装）实际上会引发异常value，但由于某种原因，已在中将其省略AccumulatorV2。