Extending Zarkov's Map/Reduce

Since I've been working on Zarkov, I've been writing a few
map/reduce jobs. One of the things I noticed about map/reduce, at
least as I've implemented it in Zarkov, is that it's pretty
inefficient if you want t o generate several aggregate views of the
same data (like say an event stream). In order to meet Zarkov's
performance requirements without making my operations team
too angry, I decided to "extend" map/reduce, as it's
implemented in Zarkov. I'm not terribly creative, so I called this
command xmapreduce. Here' I'll briefly describe
map/reduce, show the problem with it, and explain the solution
implemented in Zarkov.

Map/Reduce

The map/reduce algorithm is nice because a) it's
extremely scalable and b) it's pretty easy to grasp the basics. The
idea is that you write two functions, map and
reduce which will be applied to a large dataset to get
yourself some interesting data. For our purposes here, I'll
illustrate using Zarkov map/reduce.
Let's say you want to see how many timestamped objects in some
collection exist for each date. Here's how you'd write your Zarkov
map/reduce functions:

def map(objects): 
    for object in objects: 
        yield object['timestamp'].date(), 1 
 
def reduce(key, values): 
    return sum(values) 

The map/reduce framework will take these functions and do
(basically) the following (but efficiently distributed over several
workers):

import operator, itertools 
 
def map_reduce(input_collection, query, output_collection, map, reduce): 
    objects = input_collection.find(query) 
    map_results = list(map(objects)) 
    map_results.sort(key=operator.itemgetter(0)) 
    for key, kv_pairs in itertools.groupby(map_results, operator.itemgetter(0)): 
        value = reduce(key, [ v for k,v in kv_pairs ]) 
        output_collection.save({"_id":key, "value":value}) 

And that's pretty much all there is to it (conceptually).

The Problem with Map/Reduce

The problem arises when you have several output collections that
all depend on a single input collection (or even depend on the same
documents in the input collection). Map/reduce can get you
there, bit it'll execute the query once for each output collection,
and it needs to distribute the data to the map()
functions once for every output collection. This isn't too
efficient.
Let's extend our example above by tracking the date, month, and
year of each object. In classic map/reduce, we'd need three
map functions:

def map_date(objects): 
    for object in objects: 
        yield object['timestamp'].date(), 1 
 
def map_month(objects): 
    for object in objects: 
        yield object['timestamp'].date().replace(day=1), 1 
 
def map_year(objects): 
    for object in objects: 
        yield object['timestamp'].date().replace(month=1,day=1), 1 
 
def reduce(key, values): 
    return sum(values) 

Now, if we treat these three jobs separately, we get a lot of
duplicated effort. Particularly, Zarkov must query the input
collection three times and transfer the data to map workers three
times. (There's also duplicate serialization/deserialization of the
input objects, though the Python bson library is quite
fast.)

XMapReduce to the Rescue

The solution Zarkov uses for this case is to allow the
map function to return a target collection
along with key and value. What xmapreduce does is
transform the map functions above into one xmap
function, and tweaking our reduce function just a bit
to take the collection as an input:

def xmap(objects): 
    for object in objects: 
        yield { 
            'c':'by_date', 
            'k':object['timestamp'].date(), 
            'v':1 } 
        yield { 
            'c':'by_month', 
            'k':object['timestamp'].date().replace(day=1), 
            'v':1 } 
        yield { 
            'c':'by_year', 
            'k':object['timestamp'].date().replace(day=1, month=1), 
            'v':1 } 
 
def xreduce(collection, key, values): 
    return sum(values) 

The xmapreduce algorithm is just a slight bit more complex than
the map/reduce algorithm:

import operator, itertools 
 
def xmap_reduce(input_collection, query, output_db, map, reduce): 
    objects = input_collection.find(query) 
    map_results = list(map(objects)) 
    keyfunc = lambda doc:(doc['c'], doc['k']) 
    map_results.sort(key=keyfunc) 
    for (c,k), docs in itertools.groupby(map_results, keyfunc): 
        value = reduce(c,k, [ doc['v'] for doc in docs ])span> 
        output_db[c].save({"_id":key, "value":value}) 

Now, we can invoke a single job to calculate all three
aggregates, with close to a 3x reduction in data transfer and a
significant speedup. Assuming that a) your jobs can be
combined into a single xmapreduce job and b) you find it acceptable
to code the target collection in your map jobs, xmapreduce should
give you a significant speedup over classic map/reduce. In our
case, we started with 12 map/reduce jobs and ended up with 4
xmapreduce jobs, with a tremendous speedup, but of course your
mileage may vary. Happy hacking!