Accumulators

Overview

Learn how to use Accumulators to count events, track metrics, or debug datasets across your distributed tasks.

What You Will Learn

Detailed Concept Explanation

An Accumulator is a shared variable that worker tasks can only add to (write-only), while the driver program is the only process allowed to read its value. They are commonly used as distributed counters.

Basic Steps

1. Initialize: Create the accumulator on the driver using sc.accumulator(initialValue).
2. Update: Add to the accumulator inside worker tasks using .add(value).
3. Read: Retrieve the final aggregated sum on the driver using accum.value.

Code Examples

Input Dataset Preview

Below is the dataset containing valid records and corrupt entries:

Python (PySpark) Implementation

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("Accumulators").getOrCreate()
sc = spark.sparkContext

# Initialize counter accumulator
error_counter = sc.accumulator(0)

# Function to track corrupt lines
def parse_logs(line):
    global error_counter
    if "ERROR" in line:
        error_counter.add(1)
        return False
    return True

rdd = sc.parallelize(["valid_row_1", "ERROR: corrupt", "valid_row_2"])
valid_rdd = rdd.filter(parse_logs)

# Trigger Action (must run an action to update accumulator)
print("Valid Logs Count:", valid_rdd.count())
print("Corrupt Logs Count:", error_counter.value)

Expected Output

Valid Logs Count: 2
Corrupt Logs Count: 1

Execution Plan Diagram (Python & Scala)

Scala Implementation

import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder().appName("AccumulatorScala").getOrCreate()
val sc = spark.sparkContext

val errorCounter = sc.longAccumulator("ErrorCounter")

val rdd = sc.parallelize(Seq("valid_row_1", "ERROR: corrupt", "valid_row_2"))
val validRDD = rdd.filter(line => {
  if (line.contains("ERROR")) {
    errorCounter.add(1)
    false
  } else {
    true
  }
})

println(s"Valid Logs Count: ${validRDD.count()}")
println(s"Corrupt Logs Count: ${errorCounter.value}")

Expected Output

Valid Logs Count: 2
Corrupt Logs Count: 1

Common Mistakes

• Reading value in Worker Tasks: Trying to read accumulator.value inside a map or filter function. Worker tasks can only write (add) to accumulators; the driver is the only process allowed to read the value.
• Double Counting on Re-runs: Because of lazy evaluation, if you reference an RDD that updates an accumulator in multiple actions, Spark will recompute the RDD and add to the accumulator again, leading to double-counts. Cache the RDD to prevent this.

Best Practices

• Only Update inside Actions: Apply accumulator updates inside actions (like foreach) rather than transformations, ensuring they are only run once per record.

Interview Perspective

Interactive Challenges

Related Topics

• Broadcast Variables
• RDD Actions
• Web UI & Monitoring