Accumulator variables in Apache Spark - Practice Problems & Coding Challenges

Challenge - 5 Problems

❓ Predict Output

intermediate

Output of Spark accumulator after RDD operations

Consider the following Spark code that uses an accumulator to count even numbers in an RDD.

val accum = sc.longAccumulator("evenCount")
val rdd = sc.parallelize(1 to 5)
rdd.foreach(x => if (x % 2 == 0) accum.add(1))
println(accum.value)

What will be printed?

Apache Spark

val accum = sc.longAccumulator("evenCount")
val rdd = sc.parallelize(1 to 5)
rdd.foreach(x => if (x % 2 == 0) accum.add(1))
println(accum.value)

A 3

B 5

C 0

D 2

💡 Hint

Count how many numbers between 1 and 5 are even.

The even numbers between 1 and 5 are 2 and 4, so the accumulator counts 2.

❓ data_output

intermediate

Accumulator value after multiple actions

Given this Spark code snippet:

val accum = sc.longAccumulator("sumAccumulator")
val rdd = sc.parallelize(Seq(1, 2, 3))
rdd.foreach(x => accum.add(x))
rdd.foreach(x => accum.add(x * 2))
println(accum.value)

What is the value printed?

Apache Spark

val accum = sc.longAccumulator("sumAccumulator")
val rdd = sc.parallelize(Seq(1, 2, 3))
rdd.foreach(x => accum.add(x))
rdd.foreach(x => accum.add(x * 2))
println(accum.value)

A 12

B 18

C 21

D 9

💡 Hint

Sum all elements, then sum all elements multiplied by 2.

Sum of 1+2+3 = 6, sum of 2+4+6 = 12, total 6+12=18.

🔧 Debug

advanced

Identify the error in accumulator usage

What error will this Spark code produce?

val accum = sc.longAccumulator("accum")
val rdd = sc.parallelize(1 to 3)
val result = rdd.map(x => accum.add(x)).collect()
println(accum.value)

Apache Spark

val accum = sc.longAccumulator("accum")
val rdd = sc.parallelize(1 to 3)
val result = rdd.map(x => accum.add(x)).collect()
println(accum.value)

A 6

B Compilation error

C 0

D Runtime error: accumulator cannot be used in map

💡 Hint

Consider how accumulators behave in transformations and actions.

Accumulator updates inside map are executed on workers; collect triggers the action, so accumulator sums 1+2+3=6.

🧠 Conceptual

advanced

Why accumulators are not reliable for transformations

Why should Spark accumulators not be used to update variables inside transformations like map or filter for logic that affects output?

A Because transformations are lazy and may be recomputed, causing accumulator updates to be counted multiple times

B Because accumulators only work on the driver node and not on executors

C Because accumulators cannot be created inside transformations

D Because accumulators reset automatically after each transformation

💡 Hint

Think about Spark's lazy evaluation and task retries.

Transformations can be recomputed multiple times, so accumulator updates inside them may be counted more than once, making them unreliable for logic affecting output.

🚀 Application

expert

Using accumulators to count errors in a Spark job

You want to count how many lines in a text file contain the word "error" using Spark accumulators. Which code snippet correctly counts the occurrences?

val errorCount = sc.longAccumulator("errorCount")
val lines = sc.textFile("log.txt")
// Which option correctly updates errorCount?

Apache Spark

val errorCount = sc.longAccumulator("errorCount")
val lines = sc.textFile("log.txt")

A

val errors = lines.filter(line => line.contains("error"))
errors.foreach(_ => errorCount.add(1))
println(errorCount.value)

B

val errors = lines.filter(line => { if (line.contains("error")) errorCount.add(1); true })
errors.count()
println(errorCount.value)

C

lines.foreach(line => if (line.contains("error")) errorCount.add(1))
println(errorCount.value)

D

val errors = lines.map(line => if (line.contains("error")) errorCount.add(1))
println(errorCount.value)

💡 Hint

Remember that accumulators update reliably only in actions, not transformations.

Option C uses foreach (an action) to update the accumulator reliably. Options B and D update accumulators inside transformations or after transformations without actions, which may cause incorrect counts. Option C uses map but does not trigger an action to update the accumulator.

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