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 the 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 the 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 the Spark web UI. For all practical purposes, it is advisable 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 in Java Spark as follows:

// Imports
import org.apache.spark.sql.SparkSession;
import org.apache.spark.util.LongAccumulator;

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

// Create Accumulator variable
LongAccumulator accum = spark.sparkContext().longAccumulator("SumAccumulator");

// Add values to Accumulator
spark.sparkContext().parallelize(Arrays.asList(1, 2, 3), 1)
    .foreach(x -> accum.add(x));

// Access Accumulator value from Driver
System.out.println(accum.value());

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

Each of these accumulator classes has several methods. Among these, the add() method is called from tasks running on the cluster. Tasks cannot read the values from the accumulator; only the driver program can read the accumulator's value using the value() method.

Long Accumulator

The longAccumulator() method from the SparkContext returns LongAccumulator.

Syntax

// Long Accumulator
public LongAccumulator longAccumulator();
public LongAccumulator longAccumulator(String name);

You can create named accumulators for long type using SparkContext.longAccumulator(name).

// Imports
import org.apache.spark.sql.SparkSession;
import org.apache.spark.util.LongAccumulator;

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

// Create accumulator variable
LongAccumulator longAcc = spark.sparkContext().longAccumulator("SumAccumulator");
JavaRDD<Integer> rdd = spark.sparkContext().parallelize(Arrays.asList(1, 2, 3), 1);

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

// Access accumulator variable from Driver
System.out.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, use SparkContext.doubleAccumulator(name).

Syntax

// Double Accumulator
public DoubleAccumulator doubleAccumulator();
public DoubleAccumulator doubleAccumulator(String name);

The DoubleAccumulator class also provides methods similar to the LongAccumulator.

Collection Accumulator

For the named collection type, use SparkContext.collectionAccumulator(name).

// Collection Accumulator
public <T> CollectionAccumulator<T> collectionAccumulator();
public <T> CollectionAccumulator<T> collectionAccumulator(String name);

CollectionAccumulator class provides the following methods:

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

Note: Each of these accumulator classes has several methods. Among these, the add() method is called from tasks running on the cluster. Tasks cannot read the values from the accumulator; only the driver program can read the accumulator value using the value() method.