MapReduce: word count in three phases
Express a computation as map, shuffle, and reduce steps that can run on many machines.
from collections import Counter, defaultdict
documents = ["the cat sat", "the dog sat", "the cat ran"]
def map_phase(document):
return [(word, 1) for word in document.split()]
def shuffle(pairs):
groups = defaultdict(list)
for key, value in pairs:
groups[key].append(value)
return groups
def reduce_phase(key, values):
return key, sum(values)
pairs = [pair for document in documents for pair in map_phase(document)]
grouped = shuffle(pairs)
result = dict(reduce_phase(key, values) for key, values in sorted(grouped.items()))
print(result)
assert result == dict(Counter(" ".join(documents).split())){'cat': 2, 'dog': 1, 'ran': 1, 'sat': 2, 'the': 3}Explanation
MapReduce is a programming model for processing large data sets on clusters, introduced by Google in 2004. The programmer writes two functions. A map function takes one input record and emits key–value pairs. A reduce function takes one key and all the values collected for it and combines them. The framework handles splitting the data, scheduling work, moving data between machines, and recovering from failures.
There are three phases. In the map phase each input partition is processed independently, so many machines can work at once. In the shuffle phase the system groups all pairs by key, sending every pair with the same key to the same reducer. In the reduce phase each key's group is processed independently, again in parallel. Because tasks have no shared state, a failed task can simply be run again.
The program runs the three phases for a word count on one machine so you can see the data shapes. Each document is mapped to (word, 1) pairs, the shuffle groups the 1s by word, and the reducer sums them. The result is checked against collections.Counter. Systems such as Hadoop implement this model directly, and engines like Apache Spark generalize it with richer operations and in-memory data.
Explore the dry run
from collections import Counter, defaultdict documents = ["the cat sat", "the dog sat", "the cat ran"] def map_phase(document): return [(word, 1) for word in document.split()] def shuffle(pairs): groups = defaultdict(list) for key, value in pairs: groups[key].append(value) return groups def reduce_phase(key, values): return key, sum(values) pairs = [pair for document in documents for pair in map_phase(document)]grouped = shuffle(pairs)result = dict(reduce_phase(key, values) for key, values in sorted(grouped.items()))print(result)assert result == dict(Counter(" ".join(documents).split()))Follow the example
Start to inspect the source, values, and output. These prepared steps explain the published sample; grouped loop steps describe multiple iterations.
View the complete dry-run table
| Step / line | What happens | State after step | Output so far |
|---|---|---|---|
| Map turns one document into (word, 1) pairs, independent of every other document. | 'the cat sat' → (the,1) (cat,1) (sat,1) | — | |
| Shuffle collects every value for the same key together, as a cluster would across the network. | the → [1, 1, 1] | — | |
| Reduce combines one key's values into a result. | the → [1, 1, 1] | — | |
| The counts for all words, matching collections.Counter. | the → [1, 1, 1] | {'cat': 2, 'dog': 1, 'ran': 1, 'sat': 2, 'the': 3} |
Cost and common mistakes
Total work is O(n) for n words. On p machines the map and reduce phases ideally take O(n / p), plus the cost of the shuffle.
- Map and reduce functions must not depend on shared mutable state, otherwise tasks cannot run independently or be retried.
- A skewed key, such as a very common word, sends most data to one reducer. Combiners and partitioning strategies reduce that imbalance.
- Shuffling moves data across the network and is usually the most expensive phase.
Try it yourself
Which pairs does the reducer for the key 'sat' receive?
Reveal answer and reasoning
Two values, [1, 1], one from the second and one from the first document, and it returns ('sat', 2).