Spark自定义累加器的使用实例详解

累加器（accumulator）是Spark中提供的一种分布式的变量机制，其原理类似于mapreduce，即分布式的改变，然后聚合这些改变。累加器的一个常见用途是在调试时对作业执行过程中的事件进行计数。

累加器简单使用

Spark内置的提供了Long和Double类型的累加器。下面是一个简单的使用示例，在这个例子中我们在过滤掉RDD中奇数的同时进行计数，最后计算剩下整数的和。

val sparkConf = new SparkConf().setAppName("Test").setMaster("local[2]") 
val sc = new SparkContext(sparkConf) 
val accum = sc.longAccumulator("longAccum") //统计奇数的个数 
val sum = sc.parallelize(Array(1,2,3,4,5,6,7,8,9),2).filter(n=>{ 
 if(n%2!=0) accum.add(1L)  
 n%2==0 
}).reduce(_+_) 
println("sum: "+sum) 
println("accum: "+accum.value) 
sc.stop() 

结果为：

sum: 20
accum: 5

这是结果正常的情况，但是在使用累加器的过程中如果对于spark的执行过程理解的不够深入就会遇到两类典型的错误：少加（或者没加）、多加。

自定义累加器

自定义累加器类型的功能在1.X版本中就已经提供了，但是使用起来比较麻烦，在2.0版本后，累加器的易用性有了较大的改进，而且官方还提供了一个新的抽象类：AccumulatorV2来提供更加友好的自定义类型累加器的实现方式。官方同时给出了一个实现的示例：CollectionAccumulator类，这个类允许以集合的形式收集spark应用执行过程中的一些信息。例如，我们可以用这个类收集Spark处理数据时的一些细节，当然，由于累加器的值最终要汇聚到driver端，为了避免 driver端的outofmemory问题，需要对收集的信息的规模要加以控制，不宜过大。

继承AccumulatorV2类，并复写它的所有方法

package spark
import constant.Constant
import org.apache.spark.util.AccumulatorV2
import util.getFieldFromConcatString
import util.setFieldFromConcatString
open class SessionAccmulator : AccumulatorV2<String, String>() {
  private var result = Constant.SESSION_COUNT + "=0|"+
      Constant.TIME_PERIOD_1s_3s + "=0|"+
      Constant.TIME_PERIOD_4s_6s + "=0|"+
      Constant.TIME_PERIOD_7s_9s + "=0|"+
      Constant.TIME_PERIOD_10s_30s + "=0|"+
      Constant.TIME_PERIOD_30s_60s + "=0|"+
      Constant.TIME_PERIOD_1m_3m + "=0|"+
      Constant.TIME_PERIOD_3m_10m + "=0|"+
      Constant.TIME_PERIOD_10m_30m + "=0|"+
      Constant.TIME_PERIOD_30m + "=0|"+
      Constant.STEP_PERIOD_1_3 + "=0|"+
      Constant.STEP_PERIOD_4_6 + "=0|"+
      Constant.STEP_PERIOD_7_9 + "=0|"+
      Constant.STEP_PERIOD_10_30 + "=0|"+
      Constant.STEP_PERIOD_30_60 + "=0|"+
      Constant.STEP_PERIOD_60 + "=0"
  override fun value(): String {
    return this.result
  }
  /**
   * 合并数据
   */
  override fun merge(other: AccumulatorV2<String, String>?) {
    if (other == null) return else {
      if (other is SessionAccmulator) {
        var newResult = ""
        val resultArray = arrayOf(Constant.SESSION_COUNT,Constant.TIME_PERIOD_1s_3s, Constant.TIME_PERIOD_4s_6s, Constant.TIME_PERIOD_7s_9s,
            Constant.TIME_PERIOD_10s_30s, Constant.TIME_PERIOD_30s_60s, Constant.TIME_PERIOD_1m_3m,
            Constant.TIME_PERIOD_3m_10m, Constant.TIME_PERIOD_10m_30m, Constant.TIME_PERIOD_30m,
            Constant.STEP_PERIOD_1_3, Constant.STEP_PERIOD_4_6, Constant.STEP_PERIOD_7_9,
            Constant.STEP_PERIOD_10_30, Constant.STEP_PERIOD_30_60, Constant.STEP_PERIOD_60)
        resultArray.forEach {
          val oldValue = other.result.getFieldFromConcatString("|", it)
          if (oldValue.isNotEmpty()) {
            val newValue = oldValue.toInt() + 1
            //找到原因，一直在循环赋予值,debug30分钟 很烦
            if (newResult.isEmpty()){
              newResult = result.setFieldFromConcatString("|", it, newValue.toString())
            }
            //问题就在于这里，自定义没有写错，合并错了
            newResult = newResult.setFieldFromConcatString("|", it, newValue.toString())
          }
        }
        result = newResult
      }
    }
  }
  override fun copy(): AccumulatorV2<String, String> {
    val sessionAccmulator = SessionAccmulator()
    sessionAccmulator.result = this.result
    return sessionAccmulator
  }
  override fun add(p0: String?) {
    val v1 = this.result
    val v2 = p0
    if (v2.isNullOrEmpty()){
      return
    }else{
      var newResult = ""
      val oldValue = v1.getFieldFromConcatString("|", v2!!)
      if (oldValue.isNotEmpty()){
        val newValue = oldValue.toInt() + 1
        newResult = result.setFieldFromConcatString("|", v2, newValue.toString())
      }
      result = newResult
    }
  }
  override fun reset() {
    val newResult = Constant.SESSION_COUNT + "=0|"+
        Constant.TIME_PERIOD_1s_3s + "=0|"+
        Constant.TIME_PERIOD_4s_6s + "=0|"+
        Constant.TIME_PERIOD_7s_9s + "=0|"+
        Constant.TIME_PERIOD_10s_30s + "=0|"+
        Constant.TIME_PERIOD_30s_60s + "=0|"+
        Constant.TIME_PERIOD_1m_3m + "=0|"+
        Constant.TIME_PERIOD_3m_10m + "=0|"+
        Constant.TIME_PERIOD_10m_30m + "=0|"+
        Constant.TIME_PERIOD_30m + "=0|"+
        Constant.STEP_PERIOD_1_3 + "=0|"+
        Constant.STEP_PERIOD_4_6 + "=0|"+
        Constant.STEP_PERIOD_7_9 + "=0|"+
        Constant.STEP_PERIOD_10_30 + "=0|"+
        Constant.STEP_PERIOD_30_60 + "=0|"+
        Constant.STEP_PERIOD_60 + "=0"
    result = newResult
  }
  override fun isZero(): Boolean {
    val newResult = Constant.SESSION_COUNT + "=0|"+
        Constant.TIME_PERIOD_1s_3s + "=0|"+
        Constant.TIME_PERIOD_4s_6s + "=0|"+
        Constant.TIME_PERIOD_7s_9s + "=0|"+
        Constant.TIME_PERIOD_10s_30s + "=0|"+
        Constant.TIME_PERIOD_30s_60s + "=0|"+
        Constant.TIME_PERIOD_1m_3m + "=0|"+
        Constant.TIME_PERIOD_3m_10m + "=0|"+
        Constant.TIME_PERIOD_10m_30m + "=0|"+
        Constant.TIME_PERIOD_30m + "=0|"+
        Constant.STEP_PERIOD_1_3 + "=0|"+
        Constant.STEP_PERIOD_4_6 + "=0|"+
        Constant.STEP_PERIOD_7_9 + "=0|"+
        Constant.STEP_PERIOD_10_30 + "=0|"+
        Constant.STEP_PERIOD_30_60 + "=0|"+
        Constant.STEP_PERIOD_60 + "=0"
    return this.result == newResult
  }
}

方法介绍

value方法：获取累加器中的值

       merge方法：该方法特别重要，一定要写对，这个方法是各个task的累加器进行合并的方法（下面介绍执行流程中将要用到）

        iszero方法：判断是否为初始值

        reset方法：重置累加器中的值

        copy方法：拷贝累加器

spark中累加器的执行流程：

          首先有几个task，spark engine就调用copy方法拷贝几个累加器（不注册的），然后在各个task中进行累加（注意在此过程中，被最初注册的累加器的值是不变的），执行最后将调用merge方法和各个task的结果累计器进行合并（此时被注册的累加器是初始值）

总结

以上就是本文关于Spark自定义累加器的使用实例详解的全部内容，希望对大家有所帮助。有什么问题可以随时留言，小编会及时回复大家的。

# spark  # 自定义累加器  # 累加器  # Spark的广播变量和累加器使用方法代码示例  # Spark入门简介  # 使用docker快速搭建Spark集群的方法教程  # java 中Spark中将对象序列化存储到hdfs  # Hadoop组件简介  # 深入浅析Java Object Serialization与 Hadoop 序列化  # Java执行hadoop的基本操作实例代码  # 浅谈七种常见的Hadoop和Spark项目案例  # 自定义  # 过程中  # 是一个  # 这是  # 器中  # 有什么  # 几个  # 就会  # 是在  # 在这个  # 出了  # 在此  # 可以用  # 错了  # 有几个  # 后将  # 过大  # 类似于  # 就在于 

相关文章： 如何在阿里云域名上完成建站全流程？  焦点电影公司作品,电影焦点结局是什么？  制作网站建设的公司有哪些,网站建设比较好的公司都有哪些？  浅析上传头像示例及其注意事项  移民网站制作流程,怎么看加拿大移民官网？  早安海报制作网站推荐大全,企业早安海报怎么每天更换？  相亲简历制作网站推荐大全,新相亲大会主持人小萍萍资料？  建站之星后台密码如何安全设置与找回？  如何快速查询域名建站关键信息？  香港服务器网站测试全流程：性能评估、SEO加载与移动适配优化  ,制作一个手机app网站要多少钱？  如何在建站之星绑定自定义域名？  如何通过网站建站时间优化SEO与用户体验？  代购小票制作网站有哪些,购物小票的简要说明？  如何快速搭建自助建站会员专属系统？  电商网站制作公司有哪些,1688网是什么意思？  建站之星CMS五站合一模板配置与SEO优化指南  如何安全更换建站之星模板并保留数据？  如何通过.red域名打造高辨识度品牌网站？  手机网站制作与建设方案,手机网站如何建设？  网站代码制作软件有哪些,如何生成自己网站的代码？  极客网站有哪些,DoNews、36氪、爱范儿、虎嗅、雷锋网、极客公园这些互联网媒体网站有什么差异？  深圳网站制作的公司有哪些,dido官方网站？  IOS倒计时设置UIButton标题title的抖动问题  建站之星Pro快速搭建教程：模板选择与功能配置指南  建站之星展会模板：智能建站与自助搭建高效解决方案  上海制作企业网站有哪些,上海有哪些网站可以让企业免费发布招聘信息？  如何挑选最适合建站的高性能VPS主机？  道歉网站制作流程,世纪佳缘致歉小吴事件，相亲网站身份信息伪造该如何稽查？  宝华建站服务条款解析：五站合一功能与SEO优化设置指南  厦门模型网站设计制作公司,厦门航空飞机模型掉色怎么办？  如何高效利用亚马逊云主机搭建企业网站？  制作国外网站的软件,国外有哪些比较优质的网站推荐？  制作宣传网站的软件,小红书可以宣传网站吗？  nginx修改上传文件大小限制的方法  建站主机选虚拟主机还是云服务器更好？  制作网站的软件免费下载,免费制作app哪个平台好？  沈阳制作网站公司排名,沈阳装饰协会官方网站？  如何在企业微信快速生成手机电脑官网？  建站之星如何取消后台验证码生成？  实现虚拟支付需哪些建站技术支撑？  成都网站制作价格表,现在成都广电的单独网络宽带有多少的，资费是什么情况呢？  建站主机无法访问？如何排查域名与服务器问题  建站上传速度慢？如何优化加速网站加载效率？  建站之星伪静态规则如何正确配置？  大连企业网站制作公司,大连2025企业社保缴费网上缴费流程？  如何在万网开始建站？分步指南解析  制作门户网站的参考文献在哪,小说网站怎么建立？  如何快速搭建高效香港服务器网站？  如何基于云服务器快速搭建网站及云盘系统？