========== 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 @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 def add(a, b): 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", "waldorf salad", ] 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.compound.compound_task_execute 1 0 0 0 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 from jug.compound import CompoundTask inputs = [ "banana apple apple banana", "apple pear football", "pear", "banana apple apple", "football banana", "apple pear", "waldorf salad", ] 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) final_counts = CompoundTask(jug.mapreduce.mapreduce, merge_dicts, count1, inputs, map_step=1)