Spark自定义累加器的使用

Posted by luoxuehuan on August 7, 2016

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)
  }

}