累加器（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项目案例  # 自定义  # 过程中  # 是一个  # 这是  # 器中  # 有什么  # 几个  # 就会  # 是在  # 在这个  # 出了  # 在此  # 可以用  # 错了  # 有几个  # 后将  # 过大  # 类似于  # 就在于 

相关栏目： 【 网站优化151355 】 【 网络推广146373 】 【 网络技术251813 】 【 AI营销90571 】

相关推荐： 如何在阿里云高效完成企业建站全流程？  HTML5空格和nbsp有啥关系_nbsp的作用及使用场景【说明】  如何快速生成橙子建站落地页链接？  微信推文制作网站有哪些,怎么做微信推文，急？  Linux虚拟化技术教程_KVMQEMU虚拟机安装与调优  百度浏览器网页无法复制文字怎么办 百度浏览器复制修复  Android滚轮选择时间控件使用详解  Laravel如何为API编写文档_Laravel API文档生成与维护方法  中山网站推广排名,中山信息港登录入口？  移动端手机网站制作软件,掌上时代，移动端网站的谷歌SEO该如何做？  公司网站制作需要多少钱,找人做公司网站需要多少钱？  如何在香港服务器上快速搭建免备案网站？  如何快速启动建站代理加盟业务？  laravel怎么在请求结束后执行任务（Terminable Middleware）_laravel Terminable Middleware请求结束任务执行方法  Laravel如何安装Breeze扩展包_Laravel用户注册登录功能快速实现【流程】  bing浏览器学术搜索入口_bing学术文献检索地址  香港网站服务器数量如何影响SEO优化效果？  网站制作怎么样才能赚钱,用自己的电脑做服务器架设网站有什么利弊，能赚钱吗？  如何实现建站之星域名转发设置？  学生网站制作软件,一个12岁的学生写小说，应该去什么样的网站？  如何在万网主机上快速搭建网站？  Win10如何卸载预装Edge扩展_Win10卸载Edge扩展教程【方法】  Laravel如何处理表单验证？（Requests代码示例）  千库网官网入口推荐 千库网设计创意平台入口  无锡营销型网站制作公司,无锡网选车牌流程？  移动端脚本框架Hammer.js  Laravel怎么创建自己的包(Package)_Laravel扩展包开发入门到发布  简单实现jsp分页  如何挑选高效建站主机与优质域名？  如何在不使用负向后查找的情况下匹配特定条件前的换行符  html5怎么画眼睛_HT5用Canvas或SVG画眼球瞳孔加JS控制动态【绘制】  Linux网络带宽限制_tc配置实践解析【教程】  在线制作视频的网站有哪些,电脑如何制作视频短片？  清除minerd进程的简单方法  Laravel中的withCount方法怎么高效统计关联模型数量  javascript中对象的定义、使用以及对象和原型链操作小结  标准网站视频模板制作软件,现在有哪个网站的视频编辑素材最齐全的，背景音乐、音效等？  为什么要用作用域操作符_php中访问类常量与静态属性的优势【解答】  Laravel的辅助函数有哪些_Laravel常用Helpers函数提高开发效率  Laravel怎么定时执行任务_Laravel任务调度器Schedule配置与Cron设置【教程】  javascript日期怎么处理_如何格式化输出  广州网站制作公司哪家好一点,广州欧莱雅百库网络科技有限公司官网？  如何在沈阳梯子盘古建站优化SEO排名与功能模块？  悟空浏览器如何设置小说背景色_悟空浏览器背景色设置【方法】  Laravel如何实现数据导出到CSV文件_Laravel原生流式输出大数据量CSV【方案】  Laravel队列任务超时怎么办_Laravel Queue Timeout设置详解  Laravel如何使用模型观察者？（Observer代码示例）  Android Socket接口实现即时通讯实例代码  微博html5版本怎么弄发超话_超话进入入口及发帖格式要求【教程】  Laravel Seeder填充数据教程_Laravel模型工厂Factory使用