A process pool keeps worker processes available to execute submitted tasks. In Python, `multiprocessing.Pool` can run both stages of a MapReduce operation: map input records, group the returned values by key, then map a reducer over those groups. A pool created inside each operation is simple, but pays the process-startup cost every time. section 3.4.2
from collections import defaultdict
import multiprocessing as mp #1
def map_reduce(my_input, mapper, reducer):
with mp.Pool(2) as pool: #2
map_results = pool.map(mapper, my_input) #3
distributor = defaultdict(list)
for key, value in map_results:
distributor[key].append(value)
results = pool.map(reducer, distributor.items()) #3
return results
The two-worker implementation shows where tasks enter the pool and where grouping remains in the calling process. section 3.4.2
Keep one pool across operations when worker initialization is costly. `Pool.map` executes eagerly, unlike the lazy built-in `map`; replacing one with the other while debugging changes when work runs. Pool size also needs judgment: `os.cpu_count()` may count hyperthreads, while `len(os.sched_getaffinity())` reflects cores accessible under OS or container limits. Reduce the worker count when memory use or competing I/O would overload the machine. section 3.4.2 section 3.4.3
For per-task progress, submit work with `apply_async`, check each result’s `ready()` state, and retrieve values with `get()`. `map_async` has a callback, but it fires only when all results are ready. When finished submitting work, call `close()` and then `join()` to wait for processes to exit; `terminate()` instead forces them to stop, including unfinished work. section 3.4.3
def async_map(pool, mapper, data):
async_returns = []
for datum in data:
async_returns.append(pool.apply_async( #1
mapper, (datum, ))) #2
return async_returns
def map_reduce(pool, my_input, mapper, reducer, callback=None):
map_returns = async_map(pool, mapper, my_input)
report_progress(map_returns, 'map', callback)
map_results = [ret.get() for ret in map_returns] #3
distributor = defaultdict(list)
for key, value in map_results:
distributor[key].append(value)
returns = async_map(pool, reducer, distributor.items())
results = [ret.get() for ret in returns]
return results
Individual asynchronous results make map and reduce progress observable before an entire stage finishes. section 3.4.3
Process communication relies on pickle. A lambda used as a task function can fail to transfer to a worker; defining it with `def` resolves the demonstrated failure. File handles, database connections, sockets, and some library objects need special care or cannot be transferred. Serialization can also erase a speed gain when tasks exchange substantial data, so pools are better suited to coarse-grained work with limited communication. section 3.4.1
A pool behind an asynchronous server
An asynchronous TCP frontend can put jobs on a thread-safe work queue. A separate thread manages a persistent MapReduce pool and places completed results on another queue. In a more robust setup, pool workers ignore the interrupt signal so the main thread can handle shutdown and arrange pool cleanup. section 3.5.1 section 3.5.2
Chunk-based array processing
For a large Zarr array, a partially applied chunk-processing function can be passed to `Pool.map` with a range of chunk indices. Processing by chunk avoids concurrent writes to the same file in this example. section 8.4.4
