Spark Accumulator值未被任务读取

Question

我正在初始化一个累加器

final Accumulator<Integer> accum = sc.accumulator(0);

然后在map函数中,我试图递增累加器,然后在设置变量时使用累加器值.

JavaRDD<UserSetGet> UserProfileRDD1 = temp.map(new Function<String, UserSetGet>() {

            @Override
            public UserSetGet call(String arg0) throws Exception {

                    UserSetGet usg = new UserSetGet();

                    accum.add(1);
                    usg.setPid(accum.value().toString();


            }
  });

但我得到以下错误.

16/03/14 09:12:58 ERROR executor.Executor:阶段2.0(TID 2)中任务0.0的异常java.lang.UnsupportedOperationException:无法读取任务中的累加器值

编辑 - 根据Avihoo Mamka的回答,无法在任务中获得累加器值.

那么无论如何我可以并行实现同样的目标.这样,每当变量(例如像静态变量)在我的map函数中递增时,Pid值就会被设置.

Answer 1

来自Spark文档

累加器是仅通过关联操作"添加"的变量,因此可以并行有效地支持.它们可用于实现计数器(如MapReduce)或总和

...

只有驱动程序可以使用其value方法读取累加器的值 .

因此,当尝试从Spark中的任务中读取累加器的值时,意味着您尝试从工作程序读取其值,这与仅从驱动程序读取累加器值的概念相反.