Why is the fold action necessary in Spark?

by

Empty RDD

It cannot be substituted when RDD is empty:

val rdd = sc.emptyRDD[Int]
rdd.reduce(_ + _)
// java.lang.UnsupportedOperationException: empty collection at   
// org.apache.spark.rdd.RDD$$anonfun$reduce$1$$anonfun$apply$ ...

rdd.fold(0)(_ + _)
// Int = 0

You can of course combine reduce with condition on isEmpty but it is rather ugly.

Mutable buffer

Another use case for fold is aggregation with mutable buffer. Consider following RDD:

import breeze.linalg.DenseVector

val rdd = sc.parallelize(Array.fill(100)(DenseVector(1)), 8)

Lets say we want a sum of all elements. A naive solution is to simply reduce with +:

rdd.reduce(_ + _)

Unfortunately it creates a new vector for each element. Since object creation and subsequent garbage collection is expensive it could be better to use a mutable object. It is not possible with reduce (immutability of RDD doesn’t imply immutability of the elements), but can be achieved with fold as follows:

rdd.fold(DenseVector(0))((acc, x) => acc += x)

Zero element is used here as mutable buffer initialized once per partition leaving actual data untouched.

acc = op(obj, acc), why this operation order is used instead of acc = op(acc, obj)

See SPARK-6416 and SPARK-7683

More Related Contents:

  1. How do I split an RDD into two or more RDDs?
  2. Spark groupByKey alternative
  3. Apache Spark: What is the equivalent implementation of RDD.groupByKey() using RDD.aggregateByKey()?
  4. Apache spark dealing with case statements
  5. Spark ALS predictAll returns empty
  6. Unable to fetch the value of Println in apache spark
  7. Using a column value as a parameter to a spark DataFrame function
  8. java.lang.IllegalArgumentException at org.apache.xbean.asm5.ClassReader.(Unknown Source) with Java 10
  9. Split Spark Dataframe string column into multiple columns
  10. What does “Stage Skipped” mean in Apache Spark web UI?
  11. Avoid performance impact of a single partition mode in Spark window functions
  12. Spark performance for Scala vs Python
  13. Which operations preserve RDD order?
  14. How to split a list to multiple columns in Pyspark?
  15. pyspark dataframe filter or include based on list
  16. Spark parquet partitioning : Large number of files
  17. Number of partitions in RDD and performance in Spark
  18. Efficient pyspark join
  19. How to loop through each row of dataFrame in pyspark
  20. Fill in null with previously known good value with pyspark
  21. PySpark – get row number for each row in a group
  22. Save ML model for future usage
  23. Pyspark: Pass multiple columns in UDF
  24. Keep only duplicates from a DataFrame regarding some field
  25. reading json file in pyspark
  26. How can I access python variable in Spark SQL?
  27. PySpark: How to fillna values in dataframe for specific columns?
  28. Spark + s3 – error – java.lang.ClassNotFoundException: Class org.apache.hadoop.fs.s3a.S3AFileSystem not found
  29. Does a join of co-partitioned RDDs cause a shuffle in Apache Spark?
  30. spark.ml StringIndexer throws ‘Unseen label’ on fit()

Leave a Comment