HintsToday

🔍 What Are Accumulators in PySpark?

Accumulators are write‑only shared variables that executors can only add to, while the driver can read their aggregated value after an action completes.

🏗️ How Accumulators Are Implemented (Under the Hood)

1. Created on the driver via spark.sparkContext.longAccumulator("bad_rows").
2. Serialized and sent to each executor with the task.
3. Task‑local slot: each executor has its own copy; it calls add() without locking.
4. Task completion: executor sends the local delta back to the driver through the scheduler’s RPC.
5. Driver merges deltas using the accumulator’s merge() (for numeric types, an addition).
6. Spark UI & driver code can read the final value after the action (e.g., collect, count, write, …).

Important: If a task is retried Spark may add its contribution again, so accumulators are at‑least‑once; they should not drive program logic, only diagnostics.

🛠️ Creating & Using Accumulators

1. Built‑in LongAccumulator

bad_rows_acc = spark.sparkContext.longAccumulator("bad_rows")

def parse(line):
    try:
        return int(line)
    except ValueError:
        bad_rows_acc.add(1)
        return None        # or raise

rdd = spark.sparkContext.textFile("numbers.txt")
valid = rdd.map(parse).filter(lambda x: x is not None)
valid.sum()                # ← action triggers computation

print("Bad rows:", bad_rows_acc.value)   # safe to read here

2. Custom Accumulator (AccumulatorV2)

from pyspark.accumulators import AccumulatorParam  # <= Spark 2.0 legacy
from pyspark.sql import SparkSession
from pyspark import AccumulatorV2

class SetAccumulator(AccumulatorV2):
    def __init__(self, initial=None):
        self._set = set() if initial is None else set(initial)

    def isZero(self):
        return len(self._set) == 0

    def copy(self):
        return SetAccumulator(self._set)

    def reset(self):
        self._set.clear()

    def add(self, value):
        self._set.add(value)

    def merge(self, other):
        self._set |= other._set

    def value(self):
        return self._set

spark = SparkSession.builder.getOrCreate()
unique_errs = SetAccumulator()
spark.sparkContext.register(unique_errs, "errCodes")

# add inside tasks …

📈 Typical Use‑Cases

⚠️ Limitations & Gotchas

1. Not idempotent: task retries may duplicate additions → use only for approximate or logging stats.
2. Executors can’t read accumulator value — it’s write‑only on workers.
3. Lazy eval: value appears only after an action; reading before returns 0.
4. No control‑flow: never make program decisions based on accumulator mid‑job (could be inconsistent).
5. Streaming structured queries: not supported inside streaming map functions.

🪄 Best Practices

✅ Takeaway

Accumulators are simple, driver‑visible counters designed for instrumentation and diagnostics in PySpark:

Executors add values → Driver reads final metric.

Why do we need “special” counters at all?

Accumulators exist to collect simple metrics coming from many distributed tasks back to the driver in one aggregated value.
They solve the “many workers → one total” problem without you having to collect() the whole dataset or run a second pass with groupBy()/count().

Three ways to keep a counter in PySpark

Key distinction

Accumulators are not meant to replace normal aggregations; they are a light‑weight tap for side metrics when you don’t want to alter the main data flow.

Minimal working examples

1. Classic aggregation (accurate but full extra pass)

errors = df.filter("status = 'ERROR'").count()        # action #1
processed = df.count()                                # action #2
print(errors, processed)

Two actions → two jobs.

2. MapPartitions pattern (brings counters back in data)

def tag_and_count(iterator):
    bad = 0
    for row in iterator:
        if row.status == "ERROR":
            bad += 1
            continue
        yield row  # only good rows pass downstream
    yield {"_counter": bad}          # add one special row

cleaned = (df.rdd
           .mapPartitions(tag_and_count)
           .toDF(df.schema)          # schema must match; messy
)

Accurate, but you’ve polluted the data with “fake” rows.

3. Accumulator (side metric, one‑liner)

bad_acc = spark.sparkContext.longAccumulator("bad_rows")

def filter_good(row):
    if row.status == "ERROR":
        bad_acc.add(1)       # every executor just “fire‑and‑forgets”
        return False
    return True

good_df = df.filter(filter_good)
good_df.write.mode("overwrite").parquet("/data/good")   # single job

print("Bad rows seen:", bad_acc.value)   # safe AFTER the write

No extra shuffle, no schema tricks, counter pops up in Spark UI.

When not to use accumulators

1. Driving control flow if bad_acc.value > 1000: # ← dangerous: might still be 0 mid‑job spark.cancelJobGroup(...) The value is only reliable after the action completes.
2. Exact audit counts for financial/regulatory reporting
   Retries may double‑increment; use a deterministic aggregation instead.

When accumulators shine

Quick checklist for using accumulators safely

1. Name them: sparkContext.longAccumulator("late_events")
2. Add, don’t read, in executors
3. Read only after an action or at foreachBatch end
4. Expect slight over‑count if tasks may retry
5. Reset (acc.reset()) before the next independent job if reused

TL;DR

• Counters in distributed Spark require a mechanism that all workers can update and the driver can read.
• Accumulators give you that with almost zero coding overhead, perfect for metrics & debugging, but they’re not a substitute for real aggregations when you need 100 % accuracy.

Great question — let’s break this down carefully.

🔹 Short answer:

No, a monotonically increasing column (like monotonically_increasing_id()) is not a counter in the same sense as an accumulator.

They serve very different purposes.

🧭 Comparison: Counter vs Monotonically Increasing ID

🔍 monotonically_increasing_id() in Detail

from pyspark.sql.functions import monotonically_increasing_id

df = df.withColumn("row_id", monotonically_increasing_id())

• Spark creates a unique long number per row, often based on partition ID + record ID.
• Not guaranteed to start from 0 or be sequential: row_id: 0, 1, 4294967296, 4294967297, ...

Useful for:

• Creating row identifiers in otherwise unordered datasets.
• Doing joins where unique keys are missing.

But not useful for:

• Tallying number of errors, nulls, fallback cases
• Tracking distributed metrics
• Logging or UI metrics

🔍 Example Comparison

✅ Want to count rows with NULLs in column amount?

Accumulator:

nulls_acc = spark.sparkContext.longAccumulator("null_amounts")

df_filtered = df.filter(lambda row: not (row.amount is None or nulls_acc.add(1)))
df_filtered.write.parquet("/cleaned")
print("Rows with nulls:", nulls_acc.value)

monotonically_increasing_id():

df = df.withColumn("row_id", monotonically_increasing_id())

This helps assign IDs but doesn’t count anything.

✅ Summary Table

Pages: 1 2 3