累加器

累加器用来对信息进行聚合，通常在向 Spark传递函数时，比如使用 map() 函数或者用 filter() 传条件时，可以使用驱动器程序中定义的变量，但是集群中运行的每个任务都会得到这些变量的一份新的副本，更新这些副本的值也不会影响驱动器中的对应变量。如果我们想实现所有分片处理时更新共享变量的功能，那么累加器可以实现我们想要的效果。

1.1 系统累加器

需求：针对一个输入的日志文件，我们需要计算文件中所有空行的数量，我们可以编写以下程序

1）读取文件

scala> val notice = sc.textFile(“./NOTICE”)

notice: org.apache.spark.rdd.RDD[String] = ./NOTICE MapPartitionsRDD[40] at textFile at <console>:32

2）注册累加器

scala> val blanklines = sc.accumulator(0)

warning: there were two deprecation warnings; re-run with -deprecation for details

blanklines: org.apache.spark.Accumulator[Int] = 0

3）使用累加器，遇到空行则自增1

4）打印RDD中的数量

scala> tmp.count()

res31: Long = 3213

5）打印累加器的值,即文件中的空行数

scala> blanklines.value

res32: Int = 171

通过在驱动器中调用SparkContext.accumulator(initialValue)方法，创建出存有初始值的累加器。返回值为 org.apache.spark.Accumulator[T] 对象，其中 T 是初始值 initialValue 的类型。Spark闭包里的执行器代码可以使用累加器的 += 方法(在Java中是 add)增加累加器的值。 驱动器程序可以调用累加器的value属性(在Java中使用value()或setValue())来访问累加器的值。

注意：

（1）工作节点上的任务不能访问累加器的值。从这些任务的角度来看，累加器是一个只写变量。

（2）对于要在行动操作中使用的累加器，Spark只会把每个任务对各累加器的修改应用一次。因此，如果想要一个无论在失败还是重复计算时都绝对可靠的累加器，我们必须把它放在 foreach() 这样的行动操作中。转化操作中累加器可能会发生不止一次更新。