Explanation of the aggregate scala function

Question

I do not get to understand yet the aggregate function:

For example, having:

val x = List(1,2,3,4,5,6)
val y = x.par.aggregate((0, 0))((x, y) => (x._1 + y, x._2 + 1), (x,y) => (x._1 + y._1, x._2 + y._2))

The result will be: (21,6)

Well, I think that (x,y) => (x._1 + y._1, x._2 + y._2) is to get the result in parallel, for example it will be (1 + 2, 1 + 1) and so on.

But exactly this part that leaves me confused:

(x, y) => (x._1 + y, x._2 + 1)

why x._1 + y? and here x._2 is 0?

Thanks in advance.

Answer 1

First of all Thanks to Diego's reply which helped me connect the dots in understanding aggregate() function..

Let me confess that I couldn't sleep last night properly because I couldn't get how aggregate() works internally, I'll get good sleep tonight definitely :-)

Let's start understanding it

val result = List(1,2,3,4,5,6,7,8,9,10).par.aggregate((0, 0))
         (
          (x, y) => (x._1 + y, x._2 + 1), 
          (x,y) =>(x._1 + y._1, x._2 + y._2)
         )

result: (Int, Int) = (55,10)

aggregate function has 3 parts :

1. initial value of accumulators : tuple(0,0) here
2. seqop : It works like foldLeft with initial value of 0
3. combop : It combines the result generated through parallelization (this part was difficult for me to understand)

Let's understand all 3 parts independently :

part-1 : Initial tuple (0,0)

Aggregate() starts with initial value of accumulators x which is (0,0) here. First tuple x._1 which is initially 0 is used to compute the sum, Second tuple x._2 is used to compute total number of elements in the list.

part-2 : (x, y) => (x._1 + y, x._2 + 1)

If you know how foldLeft works in scala then it should be easy to understand this part. Above function works just like foldLeft on our List(1,2,3,4...10).

Iteration#      (x._1 + y, x._2 + 1)
     1           (0+1, 0+1)
     2           (1+2, 1+1)
     3           (3+3, 2+1)
     4           (6+4, 3+1)
     .             ....
     .             ....
     10          (45+10, 9+1)

thus after all 10 iteration you'll get the result (55,10). If you understand this part the rest is very easy but for me it was the most difficult part in understanding if all the required computation are finished then what is the use of second part i.e. compop - stay tuned :-)

part 3 : (x,y) =>(x._1 + y._1, x._2 + y._2)

Well this 3rd part is combOp which combines the result generated by different threads during parallelization, remember we used 'par' in our code to enable parallel computation of list :

List(1,2,3,4,5,6,7,8,9,10).par.aggregate(....)

Apache spark is effectively using aggregate function to do parallel computation of RDD.

Let's assume that our List(1,2,3,4,5,6,7,8,9,10) is being computed by 3 threads in parallel. Here each thread is working on partial list and then our aggregate() combOp will combine the result of each thread's computation using the below code :

(x,y) =>(x._1 + y._1, x._2 + y._2)

Original list : List(1,2,3,4,5,6,7,8,9,10)

Thread1 start computing on partial list say (1,2,3,4), Thread2 computes (5,6,7,8) and Thread3 computes partial list say (9,10)

At the end of computation, Thread-1 result will be (10,4), Thread-2 result will be (26,4) and Thread-3 result will be (19,2).

At the end of parallel computation, we'll have ((10,4),(26,4),(19,2))

Iteration#      (x._1 + y._1, x._2 + y._2)
     1           (0+10, 0+4)
     2           (10+26, 4+4)
     3           (36+19, 8+2)

which is (55,10).

Finally let me re-iterate that seqOp job is to compute the sum of all the elements of list and total number of list whereas combine function's job is to combine different partial result generated during parallelization.

I hope above explanation help you understand the aggregate().