1.为什么要使用自定义累加器
前文讲解过spark累加器的简单使用：http://blog.csdn.net/lxhandlbb/article/details/51931713
但是若业务较为复杂,需要使用多个广播变量时，就会使得程序变得非常复杂，不便于扩展维护，因此可以考虑自定义累加器。
##2.怎么使用自定义累加器
###Java版本：
package com.luoxuehuan.sparkproject.spark;
import org.apache.spark.AccumulatorParam;
import com.luoxuehuan.sparkproject.constant.Constants;
import com.luoxuehuan.sparkproject.util.StringUtils;
/**
*
* @author lxh
*
* AccumulatorParam<String>
* String 针对 String格式 进行分布式计算
* 也可以用自己的model ，但必须是可以序列化的！
* 然后基于这种特殊的数据格式，可以实现自己复杂的分布式计算逻辑
*
* 各个task 分布式在运行，可以根据你需求，task给Accumulator传入不同的值。
*
* 根据不同的值，去做复杂的逻辑。
* Spark Core里面很实用的高端技术！
*
*
*
*/
public class SessionAggrAccumulator implements AccumulatorParam<String> {
/**
*
*/
private static final long serialVersionUID = 8528303091681331462L;
/**
* Zoro方法，其实主要用于数据的初始化
* 那么，我们这里，就返回一个值，就是初始化中，所有范围区间的数量，多少0
*
* 各个范围区间的统计数量的拼接，还是采用|分割。
*/
@Override
public String zero(String v) {
return Constants.SESSION_COUNT + "=0|"
+ Constants.TIME_PERIOD_1s_3s + "=0|"
+ Constants.TIME_PERIOD_4s_6s + "=0|"
+ Constants.TIME_PERIOD_7s_9s + "=0|"
+ Constants.TIME_PERIOD_10s_30s + "=0|"
+ Constants.TIME_PERIOD_30s_60s + "=0|"
+ Constants.TIME_PERIOD_1m_3m + "=0|"
+ Constants.TIME_PERIOD_3m_10m + "=0|"
+ Constants.TIME_PERIOD_10m_30m + "=0|"
+ Constants.TIME_PERIOD_30m + "=0|"
+ Constants.STEP_PERIOD_1_3 + "=0|"
+ Constants.STEP_PERIOD_4_6 + "=0|"
+ Constants.STEP_PERIOD_7_9 + "=0|"
+ Constants.STEP_PERIOD_10_30 + "=0|"
+ Constants.STEP_PERIOD_30_60 + "=0|"
+ Constants.STEP_PERIOD_60 + "=0";
}
/**
* 这两个方法可以理解为一样的。
* 这两个方法，其实主要就是实现，v1可能就是我们初始化的那个连接串
* v2，就是我们在遍历session的时候，判断出某个session对应的区间，然后会用Constants.TIME_PERIOD_1s_3s
* 所以，我们，要做的事情就是
* 在v1中，找到v2对应的value，累加1，然后再更新回连接串里面去
*/
@Override
public String addInPlace(String v1, String v2) {
return add(v1, v2);
}
@Override
public String addAccumulator(String v1, String v2) {
return add(v1, v2);
}
/**
* session统计计算逻辑。
* @param v1 连接串
* @param v2 范围区间
* @return 更新以后的连接串
*/
private String add(String v1,String v2){
//校验：v1位空的话，直接返回v2
if(StringUtils.isEmpty(v1)) {
return v2;
}
// 使用StringUtils工具类，从v1中，提取v2对应的值，并累加1
String oldValue = StringUtils.getFieldFromConcatString(v1, "\\|", v2);
if(oldValue != null) {
// 将范围区间原有的值，累加1
int newValue = Integer.valueOf(oldValue) + 1;
// 使用StringUtils工具类，将v1中，v2对应的值，设置成新的累加后的值
return StringUtils.setFieldInConcatString(v1, "\\|", v2, String.valueOf(newValue));
}
return v1;
}
}
Scala版本：
package com.core
import com.util.{Constants, StringUtils}
import org.apache.spark.{AccumulatorParam, SparkConf, SparkContext}
/**
* Created by lxh on 2016/8/7.
*/
object SessionAggrStatAccumulatorTest {
def main(args: Array[String]) {
object SessionAggrStatAccumulator extends AccumulatorParam[String]{
/**
* 初始化和方法
* @param initialValue
* @return
*/
def zero(initialValue:String):String = {
Constants.SESSION_COUNT + "=0|" + Constants.TIME_PERIOD_1s_3s + "=0|" + Constants.TIME_PERIOD_4s_6s + "=0|" + Constants.TIME_PERIOD_7s_9s + "=0|" + Constants.TIME_PERIOD_10s_30s + "=0|" + Constants.TIME_PERIOD_30s_60s + "=0|" + Constants.TIME_PERIOD_1m_3m + "=0|" + Constants.TIME_PERIOD_3m_10m + "=0|" + Constants.TIME_PERIOD_10m_30m + "=0|" + Constants.TIME_PERIOD_30m + "=0|" + Constants.STEP_PERIOD_1_3 + "=0|" + Constants.STEP_PERIOD_4_6 + "=0|" + Constants.STEP_PERIOD_7_9 + "=0|" + Constants.STEP_PERIOD_10_30 + "=0|" + Constants.STEP_PERIOD_30_60 + "=0|" + Constants.STEP_PERIOD_60 + "=0";
}
/**
* 其次需要实现一个累加方法
* @param v1
* @param v2
* @return
*/
def addInPlace(v1:String,v2:String):String = {
/**
* 如果初始值为空,那么返回v2
*/
if(v1 ==""){
v2
}else{
/**
* 从现有的连接串中提取v2所对应的值
*/
val oldValue = StringUtils.getFieldFromConcatString(v1,"\\|",v2)
/**
* 累加1
*/
val newValue = Integer.valueOf(oldValue)+1
/**
* 改链接串中的v2设置新的累加后的值
*/
StringUtils.setFieldInConcatString(v1,"\\|",v2,String.valueOf(newValue))
}
}
}
val sparkConf = new SparkConf().setAppName("accutest").setMaster("local")
val sc = new SparkContext(sparkConf)
/**
* 使用Accumulator()()方法（curry）,创建自定义的accumulator
* 柯里化
*/
val sessionAggrStatAccumulator = sc.accumulator("")(SessionAggrStatAccumulator)
val arr = Array(Constants.TIME_PERIOD_1s_3s , Constants.TIME_PERIOD_4s_6s)
val rdd = sc.parallelize(arr,1)
rdd.foreach{sessionAggrStatAccumulator.add(_)}
println(sessionAggrStatAccumulator.value)
}
}