Accumulator variables in Apache Spark - Step-by-Step Execution

Concept Flow - Accumulator variables

Start Spark Context

↓

Create Accumulator with initial value 0

↓

Run distributed tasks

↓

Each task adds to accumulator

↓

Accumulator updates safely across nodes

↓

Collect final accumulator value

↓

Use accumulator result for analysis

Accumulator variables start at zero and safely collect sums or counts from many distributed tasks, then provide a final combined result.

Execution Sample

Apache Spark

acc = sc.accumulator(0)
rdd = sc.parallelize([1,2,3,4])
rdd.foreach(lambda x: acc.add(x))
print(acc.value)

This code creates an accumulator, adds each RDD element to it in parallel, then prints the total sum.

Execution Table

Step	Action	Accumulator Value (Local Task)	Accumulator Value (Driver)
1	Create accumulator with initial value	0	0
2	Start task 1: process element 1, add 1	1	0
3	Start task 2: process element 2, add 2	2	0
4	Start task 3: process element 3, add 3	3	0
5	Start task 4: process element 4, add 4	4	0
6	Tasks complete, driver collects updates	N/A	10
7	Print accumulator value	N/A	10

💡 All tasks finished, accumulator value collected at driver as 10 (sum of 1+2+3+4)

Variable Tracker

Variable	Start	After Task 1	After Task 2	After Task 3	After Task 4	Final
accumulator	0	1	3	6	10	10

Key Moments - 2 Insights

Why does the accumulator value on the driver stay 0 until all tasks finish?

Because each task updates its local copy of the accumulator. The driver only sees the combined result after all tasks complete, as shown in execution_table rows 2-6.

Can accumulator values be read inside tasks during execution?

No, accumulator values are only reliably read on the driver after tasks finish. Inside tasks, the accumulator is write-only, preventing inconsistent reads.

Visual Quiz - 3 Questions

Look at the execution table, what is the accumulator value on the driver after task 3 completes but before task 4?

A 3

B 0

C 6

D N/A

💡 Hint

Check the 'Accumulator Value (Driver)' column in rows 2 to 5; driver updates only after all tasks finish.

At which step does the driver finally get the total accumulator value?

A Step 4

B Step 5

C Step 6

D Step 7

💡 Hint

Look at the 'Action' column for when driver collects updates (row 6).

If the RDD had elements [1,2,3] instead, what would be the final accumulator value?

A 6

B 10

C 0

D 9

💡 Hint

Sum the elements in the RDD and compare with the final accumulator value in variable_tracker.

Concept Snapshot

Accumulator variables in Spark:
- Created with sc.accumulator(initial_value)
- Used to sum or count across distributed tasks
- Tasks add to local accumulator copies
- Driver collects final combined value after tasks finish
- Accumulators are write-only inside tasks
- Useful for debugging and metrics in distributed jobs

Full Transcript

Accumulator variables in Apache Spark start at an initial value, usually zero. When you run tasks on distributed data, each task adds to its local copy of the accumulator. These local updates do not immediately change the driver's accumulator value. After all tasks complete, Spark safely combines all local accumulator values and updates the driver's accumulator. This final value can then be read and used for analysis or debugging. Accumulators are write-only inside tasks to avoid inconsistent reads. This example showed summing numbers in an RDD using an accumulator, resulting in the total sum collected at the driver after all tasks finish.

Practice

1. What is the main purpose of an accumulator variable in Apache Spark?

easy

A. To define the schema of a DataFrame

B. To store large datasets in memory for faster access

C. To cache RDDs for repeated use

D. To count or sum values across multiple nodes during a Spark job

Solution

1. Step 1: Understand accumulator role

   Accumulators are special variables used to aggregate information like counts or sums across many worker nodes.

2. Step 2: Compare with other options

   Options B, C, and D describe caching, storage, or schema tasks, which are unrelated to accumulators.

3. Final Answer:

   To count or sum values across multiple nodes during a Spark job -> Option D

4. Quick Check:

   Accumulator = distributed counting/summing [OK]

Hint: Accumulators add values across nodes during actions [OK]

Common Mistakes:

• Confusing accumulators with broadcast variables
• Thinking accumulators cache data
• Assuming accumulators define data schema

2. Which of the following is the correct way to create an accumulator variable in Spark using Python?

easy

A. accum = sparkContext.accumulate(0)

B. accum = sparkContext.accumulator(0)

C. accum = sparkContext.accumulator()

D. accum = sparkContext.createAccumulator(0)

Solution

1. Step 1: Recall accumulator creation syntax

   The correct method to create an accumulator is sparkContext.accumulator(initial_value).

2. Step 2: Check each option

   accum = sparkContext.accumulator(0) uses the correct method and initial value. Options B, C, and D use incorrect method names or missing parameters.

3. Final Answer:

   accum = sparkContext.accumulator(0) -> Option B

4. Quick Check:

   Use .accumulator(initial_value) to create [OK]

Hint: Use .accumulator() with initial value to create [OK]

Common Mistakes:

• Using .accumulate() instead of .accumulator()
• Omitting the initial value
• Using non-existent methods like createAccumulator

3. Consider the following Spark code snippet:

accum = sc.accumulator(0)
rdd = sc.parallelize([1, 2, 3, 4])
rdd.foreach(lambda x: accum.add(x))
print(accum.value)

What will be the output?

medium

A. Error: accum.add() not allowed in transformations

B. 0

C. 10

D. 4

Solution

1. Step 1: Understand accumulator usage in foreach

   The foreach action applies the lambda to each element, adding each value to the accumulator.

2. Step 2: Calculate the sum of elements

   Sum of [1, 2, 3, 4] is 10, so accum.value will be 10 after foreach completes.

3. Final Answer:

   10 -> Option C

4. Quick Check:

   Sum of list = accumulator value = 10 [OK]

Hint: Accumulator sums values added inside actions like foreach [OK]

Common Mistakes:

• Expecting accumulator to update inside transformations
• Thinking accumulator value stays zero without action
• Confusing accumulator with variable inside map

4. What is wrong with this Spark code snippet?

accum = sc.accumulator(0)
rdd = sc.parallelize([1, 2, 3])
rdd.map(lambda x: accum.add(x))
print(accum.value)

medium

A. Accumulator updates inside map are not guaranteed to run

B. Accumulator must be initialized with a string, not 0

C. Accumulator cannot be printed directly

D. RDD must be cached before using accumulator

Solution

1. Step 1: Identify accumulator usage in map

   map is a transformation and lazy; accumulator updates inside map may not execute unless an action triggers it.

2. Step 2: Understand why accumulator value remains unchanged

   Since no action is called, accum.value remains 0 or unchanged, so this code does not update accumulator as expected.

3. Final Answer:

   Accumulator updates inside map are not guaranteed to run -> Option A

4. Quick Check:

   Accumulator updates must be inside actions [OK]

Hint: Accumulator updates only guaranteed inside actions like foreach [OK]

Common Mistakes:

• Expecting accumulator to update inside lazy transformations
• Initializing accumulator with wrong data type
• Printing accumulator without triggering action

5. You want to count how many numbers in an RDD are greater than 5 using an accumulator. Which code snippet correctly achieves this?

hard

A. accum = sc.accumulator(0)
rdd.filter(lambda x: x > 5).foreach(lambda x: accum.add(1))
print(accum.value)

B. accum = sc.accumulator(0)
rdd.map(lambda x: accum.add(1) if x > 5 else 0)
print(accum.value)

C. accum = sc.accumulator(0)
rdd.reduce(lambda x, y: accum.add(1) if x > 5 else 0)
print(accum.value)

D. accum = sc.accumulator(0)
rdd.foreach(lambda x: accum.add(1) if x > 5 else None)
print(accum.value)

Solution

1. Step 1: Identify correct accumulator update inside action

   accum = sc.accumulator(0)
   rdd.filter(lambda x: x > 5).foreach(lambda x: accum.add(1))
   print(accum.value)

   filters numbers > 5, then uses foreach (an action) to add 1 to accumulator for each such number.

2. Step 2: Check other options for errors

   accum = sc.accumulator(0)
   rdd.foreach(lambda x: accum.add(1) if x > 5 else None)
   print(accum.value)

   uses foreach with correct syntax; the original

   accum = sc.accumulator(0)
   rdd.foreach(lambda x: accum.add(1) if x > 5 else None)
   print(accum.value)

   had a syntax error missing else clause, corrected here;

   accum = sc.accumulator(0)
   rdd.map(lambda x: accum.add(1) if x > 5 else 0)
   print(accum.value)

   uses map (lazy transformation) so accumulator won't update;

   accum = sc.accumulator(0)
   rdd.reduce(lambda x, y: accum.add(1) if x > 5 else 0)
   print(accum.value)

   misuses reduce which is not for side effects.

3. Final Answer:

   accum = sc.accumulator(0) rdd.filter(lambda x: x > 5).foreach(lambda x: accum.add(1)) print(accum.value) -> Option A

4. Quick Check:

   Use filter + foreach to update accumulator correctly [OK]

Hint: Use filter then foreach to update accumulator for condition [OK]

Common Mistakes:

• Updating accumulator inside map or reduce instead of foreach
• Incorrect lambda syntax for conditional add
• Not triggering an action to update accumulator