Spark 自定义累加变量(Accmulator)AccumulatorParam
Posted FrancisYoung
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Spark 自定义累加变量(Accmulator)AccumulatorParam相关的知识,希望对你有一定的参考价值。
1.创建一个累加变量
public <T> Accumulator<T> accumulator(T initialValue, AccumulatorParam<T> param) Create an Accumulator variable of a given type, which tasks can "add" values to using the += method. Only the driver can access the accumulator‘s value. Parameters: initialValue - (undocumented) param - (undocumented) Returns: (undocumented)
使用SparkContext的如上方法,可以创建一个累加变量。默认情况下,这里的T是int或者double,因此如果想要创建T为long的累加变量是不行的。
2.AccumulatorParam介绍
概念:
initialValue:Accumulator的初始值,也就是调用SparkContext.accululator时传递的initialValue
zeroValue:AccumulatorParam的初始值,也就是zero方法的返回值。
假设样本数据集合为simple={1,2,3,4}
执行顺序:
1.调用zero(initialValue),返回zeroValue
2.调用addAccumulator(zeroValue,1) 返回v1.
调用addAccumulator(v1,2)返回v2.
调用addAccumulator(v2,3)返回v3.
调用addAccumulator(v3,4)返回v4.
3.调用addInPlace(initialValue,v4)
因此最终结果是zeroValue+1+2+3+4+initialValue.
3.实现AccumulatorParam
import org.apache.spark.AccumulatorParam; public class LongAccumulator implements AccumulatorParam<Long>{ //执行完addAccumulator方法之后,最后会执行这个方法,将value加到init。 @Override public Long addInPlace(Long init, Long value) { // TODO Auto-generated method stub // return arg0+arg1; System.out.println(init+":"+value); return init+value; } /* * init 就是SparkContext.accumulator(init)参数init。 * 这里的返回值是累计的起始值。注意哦,他可以不等于init。 * * 如果init=10,zero(init)=0,那么运算过程如下: * v1:=0+step * v1:=v1+step * ... * ... * 最后v1:=v1+init **/ @Override public Long zero(Long init) { // TODO Auto-generated method stub System.out.println(init); return 0l; } @Override public Long addAccumulator(Long value, Long step) { // TODO Auto-generated method stub System.out.println(value+","+step); return value+step; } }
接下来使用它。
import java.util.Arrays; import java.util.List; import org.apache.spark.Accumulator; import org.apache.spark.SparkConf; import org.apache.spark.api.java.JavaRDD; import org.apache.spark.api.java.JavaSparkContext; import org.apache.spark.api.java.function.VoidFunction; public class AccumulatorDemo { public static void main(String[]args){ SparkConf conf=new SparkConf().setAppName("AccumulatorDemo").setMaster("local"); JavaSparkContext sc=new JavaSparkContext(conf); Accumulator<Long> acc=sc.accumulator(0L,new LongAccumulator()); List<Long> seq=Arrays.asList(1L,2L,3L,4L); JavaRDD<Long> rdd=sc.parallelize(seq); rdd.foreach(new VoidFunction<Long>(){ @Override public void call(Long arg0) throws Exception { acc.add(arg0); } }); System.out.println(acc.value());; }
以上是关于Spark 自定义累加变量(Accmulator)AccumulatorParam的主要内容,如果未能解决你的问题,请参考以下文章