说明

分布式 共享只写变量

应用场景

在spark应用程序中，经常会有这种需求：定义一个变量，要与各个 executor 中的元素做累加运算

如：

异常监控
调试
记录符合某特性的数据的数目
在Spark Streaming应用中记录某些事件的数量

想在 Task 计算的时候统计 某些事件的数量，使用 filter 、reduce 也可以，但是使用 累加器 更方便

reduce 实现

用 reduce 累加功能

object Std1_reduce {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setMaster("local[*]").setAppName("Application")
    //构建Spark上下文对象
    val sc = new SparkContext(conf)
    val rdd = sc.makeRDD(Array(1,2,3,4,5))

    val res=rdd.reduce(_ + _)

    println(res)
    sc.stop()
  }
}

执行结果：

正确

案例

定义变量 sum，与 rdd 中的元素做运算后，查看 sum 结果

代码

import org.apache.spark.{SparkConf, SparkContext}

object Std2_变量 {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setMaster("local[*]").setAppName("Application")
    //构建Spark上下文对象
    val sc = new SparkContext(conf)
    val rdd = sc.makeRDD(Array(1,2,3,4,5,6))

    var sum = 0
    rdd.foreach(item => sum += item)

    println(sum)
    sc.stop()
  }
}

执行结果：

解释

普通变量，在分布式运行时，每个task运行的只是原始变量的副本，并不能改变原始变量的值

累加器

当这个变量被声明为 累加器 后，该变量就会有分布式计数的功能

代码

import org.apache.spark.{SparkConf, SparkContext}

object Std3_acc {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setMaster("local[*]").setAppName("Application")
    //构建Spark上下文对象
    val sc = new SparkContext(conf)
    val rdd = sc.makeRDD(Array(1,2,3,4,5))

    //使用系统默认累加器，默认初始值为0
    val sum = sc.longAccumulator("sum")
    rdd.foreach(item => sum.add(item))

    println("sum="+sum.value)

    sc.stop()
  }
}

原文出处：/show_1IX2LyGDbh0K.html