# Map/Reduce¶

## Mapping, reducing and all that Jazz¶

Jug is not a map/reduce framework. However, it is still useful to sometimes frame problems in that framework. And applying the same function to a large collection of elements (in a word, mapping) is exactly the absurdly parallel problem that Jug excels at.

### Naïve Solution¶

Let’s say you want to double all numbers between 0 and 1023. You could do this:

from jug import TaskGenerator

def double(x):
return 2*x

numbers = range(1024)
result = map(double, numbers)


This might work well for this problem. However, if instead of 1,024 numbers, you had 1 million and each computation was very fast, then this would actually be very inefficient: you are generating one task per computation. As a rule of thumb, you want your computations to last at least a few seconds, otherwise, the overhead of maintaining the infrastructure becomes too large.

### Grouping computations¶

You can use jug.mapreduce.map to achieve a better result:

from jug import mapreduce

result = mapreduce.map(double, numbers, map_step=32)


The map_step argument defines how many calls to double will be performed in a single Task.

You can also include a reduce step:

@TaskGenerator
return a + b

final = mapreduce.map(add, double, numbers, map_step=32)


this is sort of equivalent to:

final = reduce(add, map(double, numbers))


except that the order in which the reduction is done is not from left to right! In fact, this only works well if the reduction function is associative.

### Curried mapping¶

The above is fine, but sometimes you need to pass multiple arguments to the function you want to loop over:

@TaskGenerator
def distance_to(x, ref):
return abs(x - ref)

ref = 34.34
inputs = range(1024)

result = [distance_to(x, ref) for x in inputs]


This works, but we are back to where we were: too many small Tasks!

currymap to the rescue:

result = mapreduce.currymap(distance_to, [(x,ref) for x in inputs])


Arguably this function should have been called uncurrymap (as it is equivalent to the Haskell expression map . uncurry), but that just doesn’t sound right (I also like to think it’s the programming equivalent to the Currywurst, a culinary concept which almost makes me chuckle).

## Example¶

The canonical example for map/reduce is counting words in files. Here, we will do the same with some very small files:

inputs = [
"banana apple apple banana",
"apple pear football",
"pear",
"banana apple apple",
"football banana",
"apple pear",
]


The mapper function will output a dictionary of counts:

def count1(str):
from collections import defaultdict
counts = defaultdict(int)
for word in str.split():
counts[word] += 1
return counts


(We used the very useful collections.defaultdict).

While the reducer adds two dictionaries together:

def merge_dicts(rhs, lhs):
# Note that we SHOULDN'T modify arguments, so we will create a copy
rhs = rhs.copy()
for k,v in lhs.iteritems():
rhs[k] += v
return rhs


We can now use jug.mapreduce.mapreduce to put these together:

final_counts = jug.mapreduce.mapreduce(
merge_dicts,
count1,
inputs,
map_step=1)


Running jug status shows up the structure of our problem:

Task name                                    Waiting       Ready    Finished     Running
----------------------------------------------------------------------------------------
jug.mapreduce._jug_map_reduce                      0           6           0           0
jug.mapreduce._jug_reduce                          1           0           0           0
........................................................................................
Total:                                             1           6           0           0


If we had more than just 6 “files”, the values in the table would be much larger. Let’s also assume that this is part of some much larger programme that computes counts and then does some further processing with them.

Once that task is done, we might not care anymore about the break up into 6 units. So, we can wrap the whole thing into a compound task:

final_counts = CompoundTask(jug.mapreduce.mapreduce,
merge_dicts,
count1,
inputs,
map_step=1)


At first, this does not do much. The status is the same:

Task name                                    Waiting       Ready    Finished     Running
----------------------------------------------------------------------------------------
jug.mapreduce._jug_map_reduce                      0           6           0           0
jug.mapreduce._jug_reduce                          1           0           0           0
........................................................................................
Total:                                             2           6           0           0


But if we execute the tasks and re-check the status:

Task name                                    Waiting       Ready    Finished     Running
----------------------------------------------------------------------------------------
jug.mapreduce.mapreduce                            0           0           1           0
........................................................................................
Total:                                             0           0           1           0


Now, jug status reports a single task (the mapreduce task) and it is Finished.

Compound tasks not only lower the cognitive load, but they also make operations such as jug status much faster.

### Full example source code¶

We left out the imports above, but other than that, it is a fully functional example:

import jug.mapreduce

inputs = [
"banana apple apple banana",
"apple pear football",
"pear",
"banana apple apple",
"football banana",
"apple pear",
]

def count1(str):
from collections import defaultdict
counts = defaultdict(int)
for word in str.split():
counts[word] += 1
return counts

def merge_dicts(rhs, lhs):
# Note that we SHOULDN'T modify arguments, so we will create a copy
rhs = rhs.copy()
for k,v in lhs.iteritems():
rhs[k] += v
return rhs

#final_counts = jug.mapreduce.mapreduce(
#                        merge_dicts,
#                        count1,
#                        inputs,
#                        map_step=1)

merge_dicts,
count1,
inputs,
map_step=1)