MapReduce

MapReduce is a programming model and processing framework for generating and processing large datasets in parallel across a distributed cluster. The programmer writes just two functions — Map and Reduce — and the framework handles parallelisation, data distribution, load balancing, and fault tolerance.

Introduced by Google (Dean and Ghemawat, 2004), it was the processing engine that made Hadoop possible.

The Core Idea

   MAP:     take an input record, emit zero or more (key, value) pairs
            -> runs in PARALLEL on every block of the input, on the node
               that already holds that block (data locality)

   SHUFFLE & SORT:  the framework groups ALL values sharing the same key
                    and delivers them together to one reducer

   REDUCE:  take a key and the list of all its values, emit the final result
            -> runs in PARALLEL, one call per distinct key

The Classic Example — Word Count

Input (three lines, stored as three splits across the cluster):

Line 1: "the quick brown fox"
Line 2: "the lazy brown dog"
Line 3: "the fox jumps"

Phase 1 — Map

Each mapper processes one split independently, in parallel:

Mapper 1 on "the quick brown fox":
   (the, 1)  (quick, 1)  (brown, 1)  (fox, 1)

Mapper 2 on "the lazy brown dog":
   (the, 1)  (lazy, 1)  (brown, 1)  (dog, 1)

Mapper 3 on "the fox jumps":
   (the, 1)  (fox, 1)  (jumps, 1)

Phase 2 — Shuffle and Sort

The framework collects all pairs, groups by key, and sorts:

   (brown, [1, 1])
   (dog,   [1])
   (fox,   [1, 1])
   (jumps, [1])
   (lazy,  [1])
   (quick, [1])
   (the,   [1, 1, 1])

This phase is the expensive one — it moves data across the network between machines. Minimising shuffle volume is the single biggest MapReduce optimisation.

Phase 3 — Reduce

   reduce("brown", [1,1])    ->  (brown, 2)
   reduce("dog",   [1])      ->  (dog, 1)
   reduce("fox",   [1,1])    ->  (fox, 2)
   reduce("jumps", [1])      ->  (jumps, 1)
   reduce("lazy",  [1])      ->  (lazy, 1)
   reduce("quick", [1])      ->  (quick, 1)
   reduce("the",   [1,1,1])  ->  (the, 3)

Pseudocode

map(String key, String value):
    // key   = document/line identifier
    // value = the line's text
    for each word w in value:
        EmitIntermediate(w, 1)

reduce(String key, Iterator values):
    // key    = a word
    // values = list of counts for that word
    int total = 0
    for each v in values:
        total += v
    Emit(key, total)

Simulating MapReduce in Python

from collections import defaultdict

documents = [
    "the quick brown fox",
    "the lazy brown dog",
    "the fox jumps",
]

# ---------- MAP PHASE (parallel in a real cluster) ----------
def mapper(line):
    return [(word, 1) for word in line.lower().split()]

mapped = []
for doc in documents:
    mapped.extend(mapper(doc))
print("Map output:", mapped[:6], "...")

# ---------- SHUFFLE & SORT PHASE (done by the framework) ----------
grouped = defaultdict(list)
for key, value in mapped:
    grouped[key].append(value)
grouped = dict(sorted(grouped.items()))
print("\nAfter shuffle & sort:")
for k, v in grouped.items():
    print(f"  ({k}, {v})")

# ---------- REDUCE PHASE (parallel per key) ----------
def reducer(key, values):
    return (key, sum(values))

result = [reducer(k, v) for k, v in grouped.items()]
print("\nFinal output:")
for word, count in sorted(result, key=lambda x: -x[1]):
    print(f"  {word}: {count}")
# the: 3
# brown: 2
# fox: 2
# dog: 1  jumps: 1  lazy: 1  quick: 1

The Combiner — A Mini-Reducer

A combiner runs the reduce logic locally on each mapper's output, before the shuffle, to reduce network traffic.

WITHOUT combiner — mapper 1 sends 4 pairs across the network:
   (the,1) (the,1) (brown,1) (fox,1)

WITH combiner — local aggregation first, then send 3 pairs:
   (the,2) (brown,1) (fox,1)

On a real cluster with millions of records per mapper, this can cut
shuffle traffic by orders of magnitude.

Critical constraint: a combiner is only safe when the reduce operation is commutative and associative (sum, max, min, count). It is not safe for average — combining partial averages of averages gives the wrong answer. To average correctly, emit (sum, count) pairs and divide only in the final reducer.

# Demonstrating the combiner's effect
def mapper_with_combiner(line):
    local = defaultdict(int)
    for word in line.lower().split():
        local[word] += 1
    return list(local.items())

without = sum(len(mapper(d)) for d in documents)
with_comb = sum(len(mapper_with_combiner(d)) for d in documents)
print(f"Pairs shuffled without combiner: {without}")   # 11
print(f"Pairs shuffled with combiner:    {with_comb}") # 11 (small example)
# On real data with heavy key repetition, the reduction is dramatic.

# THE AVERAGE TRAP — why a naive combiner is wrong
temps = {"Delhi": [30, 32, 34], "Noida": [28, 29]}

# WRONG: average of averages
partial_avgs = [sum(v[:2])/2 for v in [temps["Delhi"]]] + [temps["Delhi"][2]]
print("Naive avg of averages:", sum(partial_avgs)/len(partial_avgs))  # 32.0 — WRONG

# CORRECT: emit (sum, count), combine those, divide at the very end
partials = [(sum(temps["Delhi"][:2]), 2), (temps["Delhi"][2], 1)]
total_sum = sum(s for s, c in partials)
total_count = sum(c for s, c in partials)
print("Correct average:", total_sum / total_count)                    # 32.0
print("Verify:", sum(temps["Delhi"]) / len(temps["Delhi"]))           # 32.0 ✓

The Partitioner

The partitioner decides which reducer receives each key:

   Default:  partition = hash(key) mod (number of reducers)

   Guarantees all values for the same key go to the SAME reducer,
   which is what makes the reduce phase correct.

A custom partitioner is used to control data distribution — for example, sending all records for one region to one reducer, or fixing data skew where one hot key (e.g. "the") overloads a single reducer while others idle.

Complete MapReduce Job Flow

Number of Mappers and Reducers

   Number of MAPPERS  = number of input splits
                      ≈ total input size / HDFS block size
                      (determined by the framework, not the programmer)

   Number of REDUCERS = set by the programmer
                        job.setNumReduceTasks(n)
                      Rule of thumb: 0.95 or 1.75 x (nodes x containers per node)

   0 reducers  ->  a MAP-ONLY job (useful for filtering/transforming;
                   map output goes straight to HDFS with no shuffle)

More Worked Examples

# EXAMPLE 2 — Maximum temperature per city per year
records = [
    ("Delhi", 2025, 42), ("Delhi", 2025, 45), ("Noida", 2025, 41),
    ("Delhi", 2026, 44), ("Noida", 2026, 43), ("Noida", 2026, 46),
    ("Delhi", 2026, 47),
]

# MAP: emit ((city, year), temperature)
mapped = [((city, year), temp) for city, year, temp in records]

# SHUFFLE & SORT
grouped = defaultdict(list)
for k, v in mapped:
    grouped[k].append(v)

# REDUCE: max per key
for key in sorted(grouped):
    print(f"{key}: max = {max(grouped[key])}")
# ('Delhi', 2025): max = 45
# ('Delhi', 2026): max = 47
# ('Noida', 2025): max = 41
# ('Noida', 2026): max = 46

# EXAMPLE 3 — Total sales per region (the canonical business query)
sales = [
    ("North", "Electronics", 1200), ("South", "Clothing", 950),
    ("North", "Clothing", 1400),    ("East",  "Books", 800),
    ("South", "Books", 1100),       ("East",  "Electronics", 700),
    ("North", "Electronics", 1350),
]

def map_sales(record):
    region, category, amount = record
    return (region, amount)

def reduce_sales(key, values):
    return (key, sum(values), len(values), round(sum(values)/len(values), 2))

mapped = [map_sales(r) for r in sales]
grouped = defaultdict(list)
for k, v in mapped:
    grouped[k].append(v)

print(f"{'Region':<8}{'Total':>8}{'Orders':>8}{'Average':>10}")
for region in sorted(grouped):
    r, total, count, avg = reduce_sales(region, grouped[region])
    print(f"{r:<8}{total:>8}{count:>8}{avg:>10}")
# North      3950       3   1316.67
# East       1500       2     750.0
# South      2050       2    1025.0

# EXAMPLE 4 — Inverted index (how search engines work)
docs = {
    "doc1": "data analytics is powerful",
    "doc2": "big data analytics with hadoop",
    "doc3": "hadoop enables big data processing",
}

# MAP: emit (word, doc_id)
mapped = [(w, d) for d, text in docs.items() for w in text.split()]

# SHUFFLE & REDUCE: collect the document list per word
index = defaultdict(set)
for word, doc in mapped:
    index[word].add(doc)

for word in sorted(index):
    print(f"{word:12} -> {sorted(index[word])}")
# analytics    -> ['doc1', 'doc2']
# big          -> ['doc2', 'doc3']
# data         -> ['doc1', 'doc2', 'doc3']
# hadoop       -> ['doc2', 'doc3']

Hadoop Streaming — MapReduce in Python

# mapper.py — reads stdin, writes tab-separated (key, value) to stdout
# import sys
# for line in sys.stdin:
#     for word in line.strip().lower().split():
#         print(f"{word}\t1")

# reducer.py — input arrives SORTED BY KEY
# import sys
# current_word, current_count = None, 0
# for line in sys.stdin:
#     word, count = line.strip().split("\t")
#     count = int(count)
#     if word == current_word:
#         current_count += count
#     else:
#         if current_word:
#             print(f"{current_word}\t{current_count}")
#         current_word, current_count = word, count
# if current_word:
#     print(f"{current_word}\t{current_count}")

# Running it on a cluster
hadoop jar /usr/lib/hadoop/hadoop-streaming.jar \
    -input /user/data/input.txt \
    -output /user/data/wordcount_out \
    -mapper "python3 mapper.py" \
    -reducer "python3 reducer.py" \
    -file mapper.py -file reducer.py

# Test locally first — this is exactly what the framework does:
# cat input.txt | python3 mapper.py | sort | python3 reducer.py

Fault Tolerance

Advantages and Limitations

MapReduce vs Spark

Spark has largely superseded MapReduce in new development, but MapReduce remains essential exam material — its model (map → shuffle → reduce) is the conceptual foundation on which Spark, Hive, and every modern distributed engine is still built.

The final lesson brings the whole course together with real-world applications of data analytics.