Core concept
val errors = spark.sparkContext.longAccumulator("errors")
df.foreach(row => if (bad(row)) errors.add(1))Advertisement
How it works
Only accumulator ops in tasks. Reads on executors return zero.
Advertisement
Trade-offs + gotchas
Debugging counters, custom metrics.