Spark Accumulators

Spark Accumulators are shared variables that are only “added” through an associative and commutative operation and are used to perform counters (Similar to Map-reduce counters) or sum operations

Spark by default supports to creation of accumulators of any numeric type and provides the capability to add custom accumulator types.

Programmers can create the following accumulators

• named accumulators
• unnamed accumulators

When you create a named accumulator, you can see it on Spark web UI under the “Accumulator” tab. On this tab, you will see two tables; 

• The first table “accumulable” – consists of all named accumulator variables and their values.  
• The second table “Tasks” – the value for each accumulator modified by a task.

Unnamed accumulators are not shown on Spark web UI, For all practical purposes, it is suggestible to use named accumulators.

Creating Accumulator variable

Spark by default provides accumulator methods for long, double, and collection types. All these methods are present in the SparkContext class and return LongAccumulator,DoubleAccumulator and  CollectionAccumulator respectively.

• LongAccumulator
• DoubleAccumulator
• CollectionAccumulator

For example, you can create a long accumulator on spark-shell

// Creating Accumulator variable
scala> val accum = sc.longAccumulator("SumAccumulator")
accum: org.apache.spark.util.LongAccumulator = LongAccumulator(id: 0, name: Some(SumAccumulator), value: 0)

The above statement creates a named accumulator “SumAccumulator”. Now, Let’s see how to add up the elements from an array to this accumulator.

scala> sc.parallelize(Array(1, 2, 3)).foreach(x => accum.add(x))
-----
-----
scala> accum.value res2: Long = 6

Each of these accumulator classes has several methods, among these, add() method calls from tasks running on the cluster. Tasks can’t read the values from the accumulator and only the driver program can read the accumulator's value using the value() method.

Long Accumulator

longAccumulator() method from SparkContext returns LongAccumulator.

Syntax

// Long Accumulator
def longAccumulator : org.apache.spark.util.LongAccumulator
def longAccumulator(name : scala.Predef.String) : org.apache.spark.util.LongAccumulator

You can create named accumulators for long type using SparkContext.longAccumulator(v) and for unnamed use signature that doesn’t take an argument.

// Imports
import org.apache.spark.sql.SparkSession

// Create SparkSession
val spark = SparkSession.builder().appName("SparkByExample")
            .master("local").getOrCreate()

// Create accumulator variable
val longAcc = spark.sparkContext.longAccumulator("SumAccumulator")    
val rdd = spark.sparkContext.parallelize(Array(1, 2, 3))

// Update accumulator variable
rdd.foreach(x => longAcc.add(x))  

// Access accumulator variable from Driver
println(longAcc.value)

The LongAccumulator class provides the following methods

• isZero
• copy
• reset
• add
• count
• sum
• avg
• merge
• value

Double Accumulator

For named double type using SparkContext.doubleAccumulator(v) and for unnamed use signature that doesn’t take an argument.

Syntax

// Double Accumulator
def doubleAccumulator : org.apache.spark.util.DoubleAccumulator
def doubleAccumulator(name : scala.Predef.String) : org.apache.spark.util.DoubleAccumulator

The DoubleAccumulator class also provides methods similar to the LongAccumulator.

Collection Accumulator

For the named collection type use SparkContext.collectionAccumulator(v) and for unnamed use signature that doesn’t take an argument.

// Collection Accumulator
def collectionAccumulator[T] : org.apache.spark.util.CollectionAccumulator[T]
def collectionAccumulator[T](name : scala.Predef.String) : org.apache.spark.util.CollectionAccumulator[T]

CollectionAccumulator class provides the following methods

• isZero
• copyAndReset
• copy
• reset
• add
• merge
• value

Note: Each of these accumulator classes has several methods, among these, add() method call from tasks running on the cluster. Tasks can’t read the values from the accumulator and only the driver program can read the accumulator value using the value() method.