pytoolz/toolz — pull request #100 by mrocklin (2013-11-15 23:23:33+00:00)
Title: WIP: Reducers and multiprocessing

This is a work in progress. Do not merge.

I want multiprocessing. There are two main impediments

1. The semantics of map, filter, reduce, etc... assume that the input and output are sequential. The Python iterator bundles up functionality (`square` in `map(square, range(10))`) to the data (`range(10)` in `map(square, range(10))`) in a way that can not be undone. Even if it does so lazily there is no easy way to pick out the `square` from the iterator after the fact.
2. The Python `multiprocessing` library uses `pickle`. The `pickle` library does not serialize functions well. In particular it doesn't handle lambdas, closures, or functions in `__main__` (i.e. interactive). These are all huge use cases for us.

My solution to 1 is to reimplement `reducers` from Clojure [1]. This bundles the `square` of `map(square, range(10))` onto a higher order function of a future reduction operation. I'm not going to type out the reasoning here. Read the blogpost below. I'm happy to explain things by chat or phone.

My solution to 2 is to leverage `dill`, an community-developed alternative to `pickle`, and `pathos` a community-developed alternative to `multiprocessing` that uses `dill`. Sadly these are not trivially pip-installable but hopefully will be soon.

In short, my goal is to be able to do something like the following for parallel wordcounting

```Python
import toolz.parallel.reducers as r
from toolz.curried import merge_with

from pathos.multiprocessing import Pool
p = Pool(8) # Our source of parallelism

filenames = ...
freq_dicts = pipe(filenames, r.map(open), r.map(concat), # many lines of text
r.map(str.split), r.map(concat), r.map(stem), # many stemmed words
r.map(frequencies)) # many frequency dictionaries

# Fold is like reduce for associative binary operators
# Here we leverage the map method from our multiprocessing pool for parallelism
result = fold(merge_with(sum), freq_dicts, {}, map=p.map)
```

This implementation cleanly separates what the computation is from how it is executed on a parallel machine. Other implementations of `map` could come in as long as they're able to serialize our generated binary operators. Naturally `builtin.map` operator works fine.

[1] http://clojure.com/blog/2012/05/08/reducers-a-library-and-model-for-collection-processing.html .