# The Aggregate Function

import findspark
findspark.init()

import pyspark

sc = pyspark.SparkContext()

# aggregate

Let’s assume an arbirtrary sequence of integers.

import numpy as np

vals = [np.random.randint(0, 10) for _ in range(20)]
vals
[5, 8, 9, 3, 0, 6, 3, 9, 8, 3, 4, 9, 5, 0, 8, 4, 2, 3, 2, 8]

rdd = sc.parallelize(vals)

### Finding the mean

Assume further that we can’t just call the handy mean method attached to our rdd object.

rdd.mean()
4.95


We’d create the mean by getting a sum of all values and a total count of numbers.

sum(vals) / len(vals)
4.95


In Spark, we recreate this logic using a two-fold reduce via a Sequence Operation and a Combination Operation.

The seqOp is a reduce step that happens per-partition. Whereas the combOp is how we take the reduced values and bring them together.

total, counts = rdd.aggregate(zeroValue=(0, 0),
seqOp=(lambda x, y: (x[0] + y, x[1] + 1)),
combOp=(lambda x, y: (x[0] + y[0], x[1] + y[1])))

total / counts
4.95


## Under the Hood

For purposes of demonstration, let’s look at something a bit easier.

Starting at 0. Simple sum. Then take the max.

rdd.aggregate(zeroValue=0,
seqOp=lambda x, y: x + y,
combOp=lambda x, y: max(x, y))
29


Why did we get this value? Peeking inside the partitions of rdd we can see the distinct groups.

brokenOut = rdd.glom().collect()
brokenOut
[[5, 8, 9, 3, 0], [6, 3, 9, 8, 3], [4, 9, 5, 0, 8], [4, 2, 3, 2, 8]]


The sum inside each partition looks like

[sum(x) for x in brokenOut]
[25, 29, 26, 19]


Thus, taking the max of each of these intermediate calculations looks like

max([sum(x) for x in brokenOut])
29


Thus, we must be careful in writing our seqOp and combOp functions, as their results depend on how the data is partitioned.