We can aggregate RDD data in Spark by using three different actions: reduce
, fold
, and aggregate
. The last one is the more general one and someway includes the first two.
As we did in our first notebook, we will use the reduced dataset (10 percent) provided for the KDD Cup 1999, containing nearly half million nework interactions. The file is provided as a Gzip file that we will download locally.
import urllib
f = urllib.urlretrieve ("http://kdd.ics.uci.edu/databases/kddcup99/kddcup.data_10_percent.gz", "kddcup.data_10_percent.gz")
data_file = "./kddcup.data_10_percent.gz"
raw_data = sc.textFile(data_file)
Both fold
and reduce
take a function as an argument that is applied to two elements of the RDD. The fold
action differs from reduce
in that it gets and additional initial zero value to be used for the initial call. This value should be the identity element for the function provided.
As an example, imagine we want to know the total duration of our interactions for normal and attack interactions. We can use reduce
as follows.
# parse data
csv_data = raw_data.map(lambda x: x.split(","))
# separate into different RDDs
normal_csv_data = csv_data.filter(lambda x: x[41]=="normal.")
attack_csv_data = csv_data.filter(lambda x: x[41]!="normal.")
The function that we pass to reduce
gets and returns elements of the same type of the RDD. If we want to sum durations we need to extract that element into a new RDD.
normal_duration_data = normal_csv_data.map(lambda x: int(x[0]))
attack_duration_data = attack_csv_data.map(lambda x: int(x[0]))
Now we can reduce these new RDDs.
total_normal_duration = normal_duration_data.reduce(lambda x, y: x + y)
total_attack_duration = attack_duration_data.reduce(lambda x, y: x + y)
print "Total duration for 'normal' interactions is {}".\
format(total_normal_duration)
print "Total duration for 'attack' interactions is {}".\
format(total_attack_duration)
Total duration for 'normal' interactions is 21075991 Total duration for 'attack' interactions is 2626792
We can go further and use counts to calculate duration means.
normal_count = normal_duration_data.count()
attack_count = attack_duration_data.count()
print "Mean duration for 'normal' interactions is {}".\
format(round(total_normal_duration/float(normal_count),3))
print "Mean duration for 'attack' interactions is {}".\
format(round(total_attack_duration/float(attack_count),3))
Mean duration for 'normal' interactions is 216.657 Mean duration for 'attack' interactions is 6.621
We have a first (and too simplistic) approach to identify attack interactions.
aggregate
¶The aggregate
action frees us from the constraint of having the return be the same type as the RDD we are working on. Like with fold
, we supply an initial zero value of the type we want to return. Then we provide two functions. The first one is used to combine the elements from our RDD with the accumulator. The second function is needed to merge two accumulators. Let's see it in action calculating the mean we did before.
normal_sum_count = normal_duration_data.aggregate(
(0,0), # the initial value
(lambda acc, value: (acc[0] + value, acc[1] + 1)), # combine value with acc
(lambda acc1, acc2: (acc1[0] + acc2[0], acc1[1] + acc2[1])) # combine accumulators
)
print "Mean duration for 'normal' interactions is {}".\
format(round(normal_sum_count[0]/float(normal_sum_count[1]),3))
Mean duration for 'normal' interactions is 216.657
In the previous aggregation, the accumulator first element keeps the total sum, while the second element keeps the count. Combining an accumulator with an RDD element consists in summing up the value and incrementing the count. Combining two accumulators requires just a pairwise sum.
We can do the same with attack type interactions.
attack_sum_count = attack_duration_data.aggregate(
(0,0), # the initial value
(lambda acc, value: (acc[0] + value, acc[1] + 1)), # combine value with acc
(lambda acc1, acc2: (acc1[0] + acc2[0], acc1[1] + acc2[1])) # combine accumulators
)
print "Mean duration for 'attack' interactions is {}".\
format(round(attack_sum_count[0]/float(attack_sum_count[1]),3))
Mean duration for 'attack' interactions is 6.621