Spark RDD Aggregate

aggregate

The aggregate function allows the user to apply two different reduce functions to the RDD. The first reduce function is applied within each partition to reduce the data within each partition into a single result. The second reduce function is used to combine the different reduced results of all partitions together to arrive at one final result. The ability to have two separate reduce functions for intra partition versus across partition reducing adds a lot of flexibility. For example the first reduce function can be the max function and the second one can be the sum function. The user also specifies an initial value. Here are some important facts.

• The initial value is applied at both levels of reduce. So both at the intra partition reduction and across partition reduction.
• Both reduce functions have to be commutative and associative.
• Do not assume any execution order for either partition computations or combining partitions.
• Why would one want to use two input data types? Let us assume we do an archaeological site survey using a metal detector. While walking through the site we take GPS coordinates of important findings based on the output of the metal detector. Later, we intend to draw an image of a map that highlights these locations using the aggregate function. In this case the zeroValue could be an area map with no highlights. The possibly huge set of input data is stored as GPS coordinates across many partitions. seqOp (first reducer) could convert the GPS coordinates to map coordinates and put a marker on the map at the respective position. combOp (second reducer) will receive these highlights as partial maps and combine them into a single final output map.

Listing Variants

def aggregate[U: ClassTag](zeroValue: U)(seqOp: (U, T) => U, combOp: (U, U) => U): U

Examples 1

val z = sc.parallelize(List(1,2,3,4,5,6), 2) // lets first print out the contents of the RDD with partition labels def myfunc(index: Int, iter: Iterator[(Int)]) : Iterator[String] = {   iter.map(x => "[partID:" +  index + ", val: " + x + "]") } z.mapPartitionsWithIndex(myfunc).collect res28: Array[String] = Array([partID:0, val: 1], [partID:0, val: 2], [partID:0, val: 3], [partID:1, val: 4], [partID:1, val: 5], [partID:1, val: 6]) z.aggregate(0)(math.max(_, _), _ + _) res40: Int = 9 // This example returns 16 since the initial value is 5 // reduce of partition 0 will be max(5, 1, 2, 3) = 5 // reduce of partition 1 will be max(5, 4, 5, 6) = 6 // final reduce across partitions will be 5 + 5 + 6 = 16 // note the final reduce include the initial value z.aggregate(5)(math.max(_, _), _ + _) res29: Int = 16 val z = sc.parallelize(List("a","b","c","d","e","f"),2) //lets first print out the contents of the RDD with partition labels def myfunc(index: Int, iter: Iterator[(String)]) : Iterator[String] = {   iter.map(x => "[partID:" +  index + ", val: " + x + "]") } z.mapPartitionsWithIndex(myfunc).collect res31: Array[String] = Array([partID:0, val: a], [partID:0, val: b], [partID:0, val: c], [partID:1, val: d], [partID:1, val: e], [partID:1, val: f]) z.aggregate("")(_ + _, _+_) res115: String = abcdef // See here how the initial value "x" is applied three times. //  - once for each partition //  - once when combining all the partitions in the second reduce function. z.aggregate("x")(_ + _, _+_) res116: String = xxdefxabc // Below are some more advanced examples. Some are quite tricky to work out. val z = sc.parallelize(List("12","23","345","4567"),2) z.aggregate("")((x,y) => math.max(x.length, y.length).toString, (x,y) => x + y) res141: String = 42 z.aggregate("")((x,y) => math.min(x.length, y.length).toString, (x,y) => x + y) res142: String = 11 val z = sc.parallelize(List("12","23","345",""),2) z.aggregate("")((x,y) => math.min(x.length, y.length).toString, (x,y) => x + y) res143: String = 10

The main issue with the code above is that the result of the inner min is a string of length 1.
The zero in the output is due to the empty string being the last string in the list. We see this result because we are not recursively reducing any further within the partition for the final string.

Examples 2

val z = sc.parallelize(List("12","23","","345"),2) z.aggregate("")((x,y) => math.min(x.length, y.length).toString, (x,y) => x + y) res144: String = 11

In contrast to the previous example, this example has the empty string at the beginning of the second partition. This results in length of zero being input to the second reduce which then upgrades it a length of 1. (Warning: The above example shows bad design since the output is dependent on the order of the data inside the partitions.)