本文主要是介绍spark的Accumulator累加器使用及 自定义Accumulator功能，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！
一、spark的累加器Accumulator
使用Accumulator时，为了保证准确性，只使用一次action操作。如果多次action操作，会造成累加器值错误。
解决方案：将任务之间的血缘依赖关系切断就可以了。什么方法有这种功能呢？cache，persist，调用这个方法的时候会将之前的依赖切除，后续的累加器就不会再被之前的transfrom操作影响到了。
二、自定义Accumulator
1. 需要重写那些方法
@Override
public String zero(String v) {}
@Override
public String addInPlace(String v1, String v2) {}
@Override
public String addAccumulator(String v1, String v2) {}
实例：
import com.yw.Util.StringUtils;
import constant.Constants;
import org.apache.spark.AccumulatorParam;
public class SessionAggrStatAccumulator implements AccumulatorParam<String> {
private static final long serialVersionUID = 6311074555136039130L;
/**
* zero方法，其实主要用于数据的初始化
* 那么，我们这里，就返回一个值，就是初始化中，所有范围区间的数量，都是0
* 各个范围区间的统计数量的拼接，还是采用一如既往的key=value|key=value的连接串的格式
*/
@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";
}
/**
* addInPlace和addAccumulator
* 可以理解为是一样的
*
* 这两个方法，其实主要就是实现，v1可能就是我们初始化的那个连接串
* v2，就是我们在遍历session的时候，判断出某个session对应的区间，
* 然后会用Constants.TIME_PERIOD_1s_3s
* 所以，我们，要做的事情就是
* 在v1中，找到v2对应的value，累加1，然后再更新回连接串里面去
*
*/
@Override
public String addInPlace(String v1, String v2) {// 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;
}
}
这篇关于spark的Accumulator累加器使用及 自定义Accumulator功能的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！
