Accumulators

Accumulators are variables that are "added" to through an associative and commutative "add" operation. They act as a container for accumulating partial values across multiple tasks (running on executors). They are designed to be used safely and efficiently in parallel and distributed Spark computations and are meant for distributed counters and sums (e.g. task metrics).

You can create built-in accumulators for longs, doubles, or collections or register custom accumulators using the SparkContext.register methods. You can create accumulators with or without a name, but only named accumulators are displayed in web UI (under Stages tab for a given stage).

spark webui accumulators.png
Figure 1. Accumulators in the Spark UI

Accumulator are write-only variables for executors. They can be added to by executors and read by the driver only.

executor1: accumulator.add(incByExecutor1)
executor2: accumulator.add(incByExecutor2)

driver:  println(accumulator.value)

Accumulators are not thread-safe. They do not really have to since the DAGScheduler.updateAccumulators method that the driver uses to update the values of accumulators after a task completes (successfully or with a failure) is only executed on a single thread that runs scheduling loop. Beside that, they are write-only data structures for workers that have their own local accumulator reference whereas accessing the value of an accumulator is only allowed by the driver.

Accumulators are serializable so they can safely be referenced in the code executed in executors and then safely send over the wire for execution.

val counter = sc.longAccumulator("counter")
sc.parallelize(1 to 9).foreach(x => counter.add(x))

Internally, longAccumulator, doubleAccumulator, and collectionAccumulator methods create the built-in typed accumulators and call SparkContext.register.

Tip
 Read the official documentation about Accumulators.
Table 1. AccumulatorV2’s Internal Registries and Counters
Name Description

metadata

AccumulatorMetadata

Used when…​FIXME

atDriverSide

Flag whether…​FIXME

Used when…​FIXME

merge Method

Caution
 FIXME

AccumulatorV2

abstract class AccumulatorV2[IN, OUT]

AccumulatorV2 parameterized class represents an accumulator that accumulates IN values to produce OUT result.

Registering Accumulator — register Method

register(
  sc: SparkContext,
  name: Option[String] = None,
  countFailedValues: Boolean = false): Unit

register creates a AccumulatorMetadata metadata object for the accumulator (with a new unique identifier) that is then used to register the accumulator with.

In the end, register registers the accumulator for cleanup (only when ContextCleaner is defined in the SparkContext).

register reports a IllegalStateException if metadata is already defined (which means that register was called already).

Cannot register an Accumulator twice.
Note
 register is a private[spark] method.
Note

register is used when:

AccumulatorMetadata

AccumulatorMetadata is a container object with the metadata of an accumulator:

  • Accumulator ID

  • (optional) name

  • Flag whether to include the latest value of an accumulator on failure

Note
 countFailedValues is used exclusively when Task collects the latest values of accumulators (irrespective of task status — a success or a failure).

Named Accumulators

An accumulator can have an optional name that you can specify when creating an accumulator.

val counter = sc.longAccumulator("counter")

AccumulableInfo

AccumulableInfo contains information about a task’s local updates to an Accumulable.

  • id of the accumulator

  • optional name of the accumulator

  • optional partial update to the accumulator from a task

  • value

  • whether or not it is internal

  • whether or not to countFailedValues to the final value of the accumulator for failed tasks

  • optional metadata

AccumulableInfo is used to transfer accumulator updates from executors to the driver every executor heartbeat or when a task finishes.

Create an representation of this with the provided values.

When are Accumulators Updated?

Examples

Example: Distributed Counter

Imagine you are requested to write a distributed counter. What do you think about the following solutions? What are the pros and cons of using it?

val ints = sc.parallelize(0 to 9, 3)

var counter = 0
ints.foreach { n =>
  println(s"int: $n")
  counter = counter + 1
}
println(s"The number of elements is $counter")

How would you go about doing the calculation using accumulators?

Example: Using Accumulators in Transformations and Guarantee Exactly-Once Update

Caution
 FIXME Code with failing transformations (tasks) that update accumulator (Map) with TaskContext info.

Example: Custom Accumulator

Caution
 FIXME Improve the earlier example

Example: Distributed Stopwatch

Note
 This is almost a raw copy of org.apache.spark.ml.util.DistributedStopwatch. 
class DistributedStopwatch(sc: SparkContext, val name: String) {

  val elapsedTime: Accumulator[Long] = sc.accumulator(0L, s"DistributedStopwatch($name)")

  override def elapsed(): Long = elapsedTime.value

  override protected def add(duration: Long): Unit = {
    elapsedTime += duration
  }
}

Further reading or watching

results matching ""

    No results matching ""