Introduction

PySpark Accumulators are shared variables that are only “added” through an associative and commutative operation and are used to perform counters (similar to MapReduce counters) or sum operations.

PySpark 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 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” – shows 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 Variables

PySpark 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 PySpark.

# Creating Accumulator variable in PySpark
from pyspark.sql import SparkSession

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

# Create accumulator variable
accum = spark.sparkContext.longAccumulator("SumAccumulator")

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

# Adding elements to accumulator
rdd = spark.sparkContext.parallelize([1, 2, 3])
rdd.foreach(lambda x: accum.add(x))

# Access accumulator value
print(accum.value)

Each of these accumulator classes has several methods, among these, add() method is called 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

The longAccumulator() method from SparkContext returns a LongAccumulator.

Example

# Long Accumulator in PySpark
long_acc = spark.sparkContext.longAccumulator("SumAccumulator")
rdd = spark.sparkContext.parallelize([1, 2, 3])

# Update accumulator variable
rdd.foreach(lambda x: long_acc.add(x))

# Access accumulator variable from Driver
print(long_acc.value)

The LongAccumulator class provides the following methods:

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

Double Accumulator

You can create named double accumulators using SparkContext.doubleAccumulator(v). To create an unnamed one, use the signature that doesn’t take an argument.

Example

# Double Accumulator in PySpark
double_acc = spark.sparkContext.doubleAccumulator("DoubleSumAccumulator")
rdd = spark.sparkContext.parallelize([1.0, 2.0, 3.0])

# Update accumulator variable
rdd.foreach(lambda x: double_acc.add(x))

# Access accumulator
print(double_acc.value)

The DoubleAccumulator class also provides methods similar to the LongAccumulator.

Collection Accumulator

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

Example

# Collection Accumulator in PySpark
collection_acc = spark.sparkContext.collectionAccumulator("CollectionAccumulator")
rdd = spark.sparkContext.parallelize([1, 2, 3])

# Update accumulator
rdd.foreach(lambda x: collection_acc.add(x))

# Access collection
print(collection_acc.value)

The 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 can’t read the values from the accumulator, and only the driver program can read the accumulator's value using the value() method.