How to compute cumulative sum using Spark

by

  1. Compute partial results for each partition:

    val partials = rdd.mapPartitionsWithIndex((i, iter) => {
  val (keys, values) = iter.toSeq.unzip
  val sums  = values.scanLeft(0)(_ + _)
  Iterator((keys.zip(sums.tail), sums.last))
})

  2. Collect partials sums 

    val partialSums = partials.values.collect

  3. Compute cumulative sum over partitions and broadcast it:

    val sumMap = sc.broadcast(
  (0 until rdd.partitions.size)
    .zip(partialSums.scanLeft(0)(_ + _))
    .toMap
)

  4. Compute final results:

    val result = partials.keys.mapPartitionsWithIndex((i, iter) => {
  val offset = sumMap.value(i)
  if (iter.isEmpty) Iterator()
  else iter.next.map{case (k, v) => (k, v + offset)}.toIterator
})

More Related Contents:

  1. Spark Scala: How to convert Dataframe[vector] to DataFrame[f1:Double, …, fn: Double)]
  2. How to define schema for custom type in Spark SQL?
  3. Specifying the filename when saving a DataFrame as a CSV [duplicate]
  4. How can I change column types in Spark SQL’s DataFrame?
  5. NullPointerException in Scala Spark, appears to be caused be collection type?
  6. What does setMaster `local[*]` mean in spark?
  7. Spark unionAll multiple dataframes
  8. How can I pass extra parameters to UDFs in Spark SQL?
  9. How to split a dataframe into dataframes with same column values?
  10. Spark performance for Scala vs Python
  11. How to use regex to include/exclude some input files in sc.textFile?
  12. How to define a custom aggregation function to sum a column of Vectors?
  13. Spark losing println() on stdout
  14. How to write spark streaming DF to Kafka topic
  15. Spark DataFrame: does groupBy after orderBy maintain that order?
  16. Stackoverflow due to long RDD Lineage
  17. Better way to convert a string field into timestamp in Spark
  18. Spark UDAF with ArrayType as bufferSchema performance issues
  19. How to find spark RDD/Dataframe size?
  20. How to get ID of a map task in Spark?
  21. Spark: produce RDD[(X, X)] of all possible combinations from RDD[X]
  22. Provide schema while reading csv file as a dataframe in Scala Spark
  23. Apache Spark, add an “CASE WHEN … ELSE …” calculated column to an existing DataFrame
  24. Calculate Cosine Similarity Spark Dataframe
  25. How to sort an RDD in Scala Spark?
  26. Difference between two rows in Spark dataframe
  27. Run a scala code jar appear NoSuchMethodError:scala.Predef$.refArrayOps
  28. Spark – Error “A master URL must be set in your configuration” when submitting an app
  29. Spark UDF with varargs
  30. How to transpose an RDD in Spark

Leave a Comment