When are accumulators truly reliable?

by

To answer the question “When are accumulators truly reliable ?”

Answer : When they are present in an Action operation.

As per the documentation in Action Task, even if any restarted tasks are present it will update Accumulator only once.

For accumulator updates performed inside actions only, Spark guarantees that each task’s update to the accumulator will only be applied once, i.e. restarted tasks will not update the value. In transformations, users should be aware of that each task’s update may be applied more than once if tasks or job stages are re-executed.

And Action do allow to run custom code.

For Ex.

val accNotEmpty = sc.accumulator(0)
ip.foreach(x=>{
  if(x!=""){
    accNotEmpty += 1
  }
})

But, Why Map+Action viz. Result Task operations are not reliable for an Accumulator operation?

  1. Task failed due to some exception in code. Spark will try 4 times(default number of tries).If task fail every time it will give an exception.If by chance it succeeds then Spark will continue and just update the accumulator value for successful state and failed states accumulator values are ignored.
    Verdict : Handled Properly
  2. Stage Failure : If an executor node crashes, no fault of user but an hardware failure – And if the node goes down in shuffle stage.As shuffle output is stored locally, if a node goes down, that shuffle output is gone.So Spark goes back to the stage that generated the shuffle output, looks at which tasks need to be rerun, and executes them on one of the nodes that is still alive.After we regenerate the missing shuffle output, the stage which generated the map output has executed some of it’s tasks multiple times.Spark counts accumulator updates from all of them.
    Verdict : Not handled in Result Task.Accumulator will give wrong output.
  3. If a task is running slow then, Spark can launch a speculative copy of that task on another node.
    Verdict : Not handled.Accumulator will give wrong output.
  4. RDD which is cached is huge and can’t reside in Memory.So whenever the RDD is used it will re run the Map operation to get the RDD and again accumulator will be updated by it.
    Verdict : Not handled.Accumulator will give wrong output.

So it may happen same function may run multiple time on same data.So Spark does not provide any guarantee for accumulator getting updated because of the Map operation.

So it is better to use Accumulator in Action operation in Spark.

To know more about Accumulator and its issues refer this Blog Post – By Imran Rashid.

More Related Contents:

  1. How to connect Spark SQL to remote Hive metastore (via thrift protocol) with no hive-site.xml?
  2. Spark SQL – load data with JDBC using SQL statement, not table name
  3. Spark Dataframe validating column names for parquet writes
  4. Spark: subtract two DataFrames
  5. How to improve performance for slow Spark jobs using DataFrame and JDBC connection?
  6. TypeError: Column is not iterable – How to iterate over ArrayType()?
  7. How to overwrite the output directory in spark
  8. Is groupByKey ever preferred over reduceByKey
  9. Filtering a spark dataframe based on date
  10. Pyspark : forward fill with last observation for a DataFrame
  11. spark.sql.crossJoin.enabled for Spark 2.x
  12. Why do I have to explicitly tell Spark what to cache?
  13. multiple conditions for filter in spark data frames
  14. Why does Spark think this is a cross / Cartesian join
  15. How to pass whole Row to UDF – Spark DataFrame filter
  16. Rename more than one column using withColumnRenamed
  17. Saving dataframe to local file system results in empty results
  18. Dealing with a large gzipped file in Spark
  19. What should be the optimal value for spark.sql.shuffle.partitions or how do we increase partitions when using Spark SQL?
  20. pyspark: Efficiently have partitionBy write to same number of total partitions as original table
  21. Spark lists all leaf node even in partitioned data
  22. AWS EMR – ModuleNotFoundError: No module named ‘pyarrow’
  23. Array Intersection in Spark SQL
  24. What do the numbers on the progress bar mean in spark-shell?
  25. pyspark: rolling average using timeseries data
  26. How to improve broadcast Join speed with between condition in Spark
  27. Spark ALS predictAll returns empty
  28. Spark SQL referencing attributes of UDT
  29. Why is the fold action necessary in Spark?
  30. How to add a SparkListener from pySpark in Python?

Leave a Comment