累加器（accumulator）是Spark中提供的一種分散式的變數機制，其原理類似於mapreduce，即分散式的改變，然後聚合這些改變。累加器的一個常見用途是在除錯時對作業執行過程中的事件進行計數。

累加器簡單使用

Spark內建的提供了Long和Double型別的累加器。下面是一個簡單的使用示例，在這個例子中我們在過濾掉RDD中奇數的同時進行計數，最後計算剩下整數的和。

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

sum: 20
accum: 5

這是結果正常的情況，但是在使用累加器的過程中如果對於spark的執行過程理解的不夠深入就會遇到兩類典型的錯誤：少加（或者沒加）、多加。

少加的情況：

對於如下程式碼：

1. val accum = sc.longAccumulator( "longAccum")
2. val numberRDD = sc.parallelize(Array( 1, 2, 3, 4, 5, 6, 7, 8, 9), 2).map(n=>{
3. accum.add( 1L)
4. n+ 1
5. })
6. println( "accum: "+accum.value)

多加的情況：

對於如下程式碼：

1. val accum = sc.longAccumulator( "longAccum")
2. val numberRDD = sc.parallelize(Array( 1, 2, 3, 4, 5, 6, 7, 8, 9), 2).map(n=>{
3. accum.add( 1L)
4. n+ 1
5. })
6. numberRDD. count
7. println ("accum1:"+accum.value)
8. numberRDD. reduce (_+_)
9. println ("accum2: "+accum.value)

accum1:9

accum2: 18

我們雖然只在map裡進行了累加器加1的操作，但是兩次得到的累加器的值卻不一樣，這是由於count和reduce都是action型別的操作，觸發了兩次作業的提交，所以map運算元實際上被執行了了兩次，在reduce操作提交作業後累加器又完成了一輪計數，所以最終累加器的值為18。究其原因是因為count雖然促使numberRDD被計出來，但是由於沒有對其進行快取，所以下次再次需要使用numberRDD這個資料集是，還需要從並行化資料集的部分開始執行計算。解釋到這裡，這個問題的解決方法也就很清楚了，就是在count之前呼叫numberRDD的cache方法（或persist），這樣在count後資料集就會被快取下來，reduce操作就會讀取快取的資料集而無需從頭開始計算了。改成如下程式碼即可：

1. val accum = sc.longAccumulator( "longAccum")
2. val numberRDD = sc.parallelize(Array( 1, 2, 3, 4, 5, 6, 7, 8, 9), 2).map(n=>{
3. accum.add( 1L)
4. n+ 1
5. })
6. numberRDD.cache(). count
7. println ("accum1:"+accum.value)
8. numberRDD. reduce (_+_)
9. println ("accum2: "+accum.value)

自定義累加器

1. import java.util
3. import org.apache.spark.util.AccumulatorV2
5. class LogAccumulator extends AccumulatorV2[String, java.util.Set[String]] {
6. private val _logArray: java.util.Set[String] = new java.util.HashSet[String]()
8. override def isZero: Boolean = {
9. _logArray.isEmpty
10. }
12. override def reset(): Unit = {
13. _logArray.clear()
14. }
16. override def add(v: String): Unit = {
17. _logArray.add(v)
18. }
20. override def merge(other: AccumulatorV2[String, java.util.Set[String]]): Unit = {
21. other match {
22. case o: LogAccumulator => _logArray.addAll(o.value)
23. }
25. }
27. override def value: java.util.Set[String] = {
28. java.util.Collections.unmodifiableSet(_logArray)
29. }
31. override def copy(): AccumulatorV2[String, util.Set[String]] = {
32. val newAcc = new LogAccumulator()
33. _logArray. synchronized{
34. newAcc._logArray.addAll(_logArray)
35. }
36. newAcc
37. }
38. }

1. import scala.collection.JavaConversions._
3. import org.apache.spark.{SparkConf, SparkContext}
5. object Main {
6. def main(args: Array[String]): Unit = {
7. val sparkConf = new SparkConf().setAppName( "Test").setMaster( "local[2]")
8. val sc = new SparkContext(sparkConf)
9. val accum = new LogAccumulator
10. sc.register(accum, "logAccum")
11. val sum = sc.parallelize(Array( "1", "2a", "3", "4b", "5", "6", "7cd", "8", "9"), 2).filter(line => {
12. val pattern = "" "^-?(\d+)" ""
13. val flag = line.matches(pattern)
14. if (!flag) {
15. accum.add(line)
16. }
17. flag
18. }).map(_.toInt).reduce(_ + _)
20. println( "sum: " + sum)
21. for (v <- accum.value) print(v + " ")
22. println()
23. sc.stop()
24. }
25. }

來源：https://blog.csdn.net/u013468917/article/details/70617085