This exercise implements a complete MapReduce pipeline in pure Python, then demonstrates how it maps to Hadoop Streaming for execution on an HDFS cluster. Rather than requiring a live Hadoop cluster, the exercise runs the full map-shuffle-reduce logic locally against a simulated IPL commentary dataset, producing the identical output that a Hadoop Streaming job would produce on the same data. The local simulation exercises every phase — mapper, combiner, shuffle, reducer — and validates correctness with explicit assertions before the Hadoop Streaming version is shown.
The dataset is IPL ball-by-ball commentary lines — one line per delivery — and the task is a multi-key frequency count: count total deliveries, boundaries, and wickets per bowler across all matches. This goes beyond the standard word count to demonstrate that the MapReduce model handles any grouping-and-aggregation problem, not just word frequency. The exercise ends with a data locality and block placement simulation that illustrates how YARN would schedule mappers across DataNodes for this input.
Analogy🏏Cricket
🏏 Think of it like cricket: Imagine the DRS review system deployed across three independent video-review centres in Mumbai, Chennai, and London, each holding a copy of the ball-tracking data. A CAP partition is a network outage that cuts communication between them. A CP system says: if the centres cannot synchronise, no review decision is issued — no player walks until the system is restored. Consistency is guaranteed; availability is sacrificed. An AP system says: each centre issues its own decision based on its local data, even if that means Mumbai says out and London says not out — reviews continue but different centres may give different verdicts. Partition tolerance is non-negotiable because the network always has the possibility of failing; the choice is whether umpires wait for consensus or proceed with local data.
Step 1 — Data Preparation
Generate the synthetic IPL commentary dataset representing three HDFS input blocks — one per match — and write each block to a separate file simulating how HDFS would split the input. Each line is a tab-separated delivery record. Verify the total line count and distribution across blocks before running the MapReduce pipeline, since uneven block distribution is a common source of skew in real cluster jobs.
Analogy🏏Cricket
🏏 Think of it like cricket: splitting the input into three block files is like a season where three separate scorers each keep the book for one match — the league never keeps one giant scroll; the record is naturally divided into per-match scorebooks, just as HDFS splits a large file into blocks stored on different nodes. Verifying the total line count before processing is the scorers' reconciliation ritual: every over must show exactly six legal deliveries, and the three books together must account for all 360 balls bowled. And just as a double-header would leave one scorer with twice the entries of the others — slowing the whole compilation down because everyone waits for the busiest scorer — an uneven block distribution creates skew, where one mapper becomes the straggler that delays the entire job. Checking the distribution up front, like checking each book's tally before the stats meeting, catches the imbalance before it costs you the whole evening.
python
# exercise_mapreduce.py — Step 1: Data preparation
import csv
import io
import random
from pathlib import Path
random.seed(2024)
BOWLERS = ["Bumrah", "Shami", "Ashwin", "Jadeja", "Hardik", "Siraj", "Chahal"]
BATTERS = ["Rohit", "Kohli", "Gill", "Dhoni", "Warner", "Maxwell", "Stokes"]
TEAMS = ["Mumbai Indians", "CSK", "RCB", "KKR"]
def generate_match_block(match_id: int, n_deliveries: int = 120) -> list[str]:
"""
Generate n_deliveries commentary lines for one match.
Format: delivery_id\tmatch_id\tbowler\tbatter\truns\tis_wicket
"""
lines = []
for i in range(n_deliveries):
runs = random.choices([0,1,2,4,6], weights=[35,30,15,15,5])[0]
is_wicket = random.random() < 0.03
lines.append(
f"{match_id*1000+i}\t{match_id}\t{random.choice(BOWLERS)}\t"
f"{random.choice(BATTERS)}\t{runs}\t{1 if is_wicket else 0}\n"
)
return lines
# Generate three HDFS blocks (one per match file)
block_dir = Path("hdfs_blocks")
block_dir.mkdir(exist_ok=True)
total_lines = 0
for match_id in [10001, 10002, 10003]:
lines = generate_match_block(match_id, 120)
block_file = block_dir / f"match_{match_id}.tsv"
block_file.write_text("".join(lines))
total_lines += len(lines)
print(f"Block written: {block_file} ({len(lines)} deliveries)")
print(f"\nTotal input lines across all blocks: {total_lines}")
assert total_lines == 360, f"Expected 360 deliveries, got {total_lines}"
print("Step 1 complete — input blocks created")Step 2 — Mapper, Combiner, and Shuffle
Implement the mapper that reads delivery lines and emits `(bowler, stats_dict)` pairs, the combiner that pre-aggregates stats within each block to reduce shuffle volume, and the shuffle function that groups all values by bowler key. Measure the shuffle reduction factor — the ratio of combiner output pairs to mapper output pairs — to quantify the combiner's benefit. Verify that the shuffle groups every bowler's data together with no mixing of keys across groups.
Analogy🏏Cricket
🏏 Think of it like cricket: the mapper is each match's scorer reading their own book ball by ball and writing one slip per delivery — 'Bumrah: 1 delivery, 0 boundaries, 1 wicket'. Mailing all 120 raw slips per match to the league office would flood the post. So the combiner acts like the scorer compiling a bowling card at the close of their own match: Bumrah's 24 slips collapse into one line of match figures before anything leaves the ground. The shuffle is the league office sorting the incoming mail into one pigeonhole per bowler, so every card for the same bowler — from all three grounds — lands in the same pile. The shuffle reduction factor is literally the postage saved: cards posted divided by slips written. Just as pre-tallying at the ground cuts what travels between venues, the combiner cuts network transfer between mapper and reducer — and the grouping guarantee means the season statistician never has to hunt across piles for a stray card.
python
# exercise_mapreduce.py — Step 2: Mapper, combiner, shuffle
from collections import defaultdict
from pathlib import Path
def mapper(line: str) -> tuple[str, dict] | None:
"""
Map one delivery line to (bowler, {deliveries, runs, wickets, boundaries}).
Returns None for malformed lines.
"""
parts = line.strip().split("\t")
if len(parts) < 6:
return None
try:
bowler = parts[2]
runs = int(parts[4])
is_wicket = int(parts[5])
except (ValueError, IndexError):
return None
return (
bowler,
{
"deliveries": 1,
"runs": runs,
"wickets": is_wicket,
"boundaries": 1 if runs >= 4 else 0,
}
)
def combiner(pairs: list[tuple]) -> list[tuple]:
"""Pre-aggregate within one mapper block to reduce shuffle volume."""
combined: dict[str, dict] = defaultdict(
lambda: {"deliveries": 0, "runs": 0, "wickets": 0, "boundaries": 0}
)
for key, val in pairs:
for field in ["deliveries", "runs", "wickets", "boundaries"]:
combined[key][field] += val[field]
return list(combined.items())
def shuffle(all_pairs: list[tuple]) -> dict[str, list]:
"""Sort and group all intermediate pairs by key — simulates YARN shuffle."""
grouped: dict[str, list] = defaultdict(list)
for key, val in sorted(all_pairs, key=lambda x: x[0]): # sort by key
grouped[key].append(val)
return dict(grouped)
# Run Map phase on all three blocks
all_map_output = []
for block_file in sorted(Path("hdfs_blocks").glob("*.tsv")):
lines = block_file.read_text().splitlines()
map_pairs = [mapper(line) for line in lines]
map_pairs = [p for p in map_pairs if p is not None] # filter bad lines
print(f"Block {block_file.name}: {len(map_pairs)} mapper output pairs")
# Apply combiner before adding to shuffle input
combined = combiner(map_pairs)
print(f" After combiner: {len(combined)} pairs "
f"(reduction factor: {len(map_pairs)/len(combined):.1f}x)")
all_map_output.extend(combined)
# Shuffle phase: group by bowler key
grouped = shuffle(all_map_output)
print(f"\nShuffle complete: {len(grouped)} distinct bowler keys")
for bowler, vals in grouped.items():
print(f" {bowler:<10}: {len(vals)} partial records to reduce")
# Verify: every bowler appears exactly once as a key
assert len(grouped) <= len(set(BOWLERS))
print("Step 2 complete — shuffle correct")Step 3 — Reducer and Output Validation
Implement the reducer that aggregates all partial stats for each bowler into final season totals and computes economy rate, then run it across all shuffled groups. Write the results to a TSV output file simulating HDFS output. Validate the results: total deliveries across all bowlers must equal 360 (the full input), and economy values must be between 0 and 36 (theoretical bounds for a bowler who concedes 36 runs per over). Assert both conditions before marking the step complete.
Analogy🏏Cricket
🏏 Think of it like cricket: the reducer is the season statistician working through one pigeonhole at a time: take Bumrah's three match cards, add the deliveries, boundaries and wickets, and compute his season economy — one final line per bowler, exactly as the reducer emits one output record per key. Writing the TSV output is publishing the official season bowling table. The validation step is the scorer's balancing discipline every scorebook demands: just as runs off the bat plus extras must equal the team total, the deliveries summed across all bowlers must equal the 360 balls actually bowled — if a single slip went missing in the shuffle, the books will not balance and you know immediately. The economy bound check works like knowing 36 is the most a bowler can concede off a legal over (six sixes): any figure outside 0–36 is not a surprising performance, it is a corrupted record. Balanced books are the payoff — output you can publish without re-checking a single ball.
python
# exercise_mapreduce.py — Step 3: Reducer, output, validation
import csv
from pathlib import Path
def reducer(bowler: str, partial_stats: list[dict]) -> dict:
"""Aggregate all partial stats for one bowler into final season totals."""
total = {"deliveries": 0, "runs": 0, "wickets": 0, "boundaries": 0}
for s in partial_stats:
for field in total:
total[field] += s[field]
overs = total["deliveries"] / 6
economy = round(total["runs"] / overs, 2) if overs > 0 else 0.0
sr = round((total["boundaries"] / total["deliveries"]) * 100, 1) \
if total["deliveries"] > 0 else 0.0
return {
"bowler": bowler,
"deliveries": total["deliveries"],
"overs": round(overs, 1),
"runs": total["runs"],
"wickets": total["wickets"],
"boundaries": total["boundaries"],
"economy": economy,
"boundary_pct":sr,
}
# Run Reduce phase
results = [reducer(bowler, vals) for bowler, vals in grouped.items()]
results.sort(key=lambda x: x["economy"])
# Write to output file (simulates HDFS output part-00000)
output_path = Path("hdfs_output/part-00000.tsv")
output_path.parent.mkdir(exist_ok=True)
fieldnames = ["bowler","deliveries","overs","runs","wickets","economy","boundary_pct"]
with output_path.open("w", newline="") as f:
writer = csv.DictWriter(f, fieldnames=fieldnames, delimiter="\t",
extrasaction="ignore")
writer.writeheader()
writer.writerows(results)
print("MapReduce output:")
print(f" {'Bowler':<10} {'Deliveries':>10} {'Overs':>6} {'Runs':>6} "
f"{'Wickets':>8} {'Economy':>8} {'Boundary%':>10}")
print(" " + "-"*62)
for r in results:
print(f" {r['bowler']:<10} {r['deliveries']:>10} {r['overs']:>6} "
f"{r['runs']:>6} {r['wickets']:>8} {r['economy']:>8} {r['boundary_pct']:>10}")
# Validation assertions
total_deliveries = sum(r["deliveries"] for r in results)
assert total_deliveries == 360, f"Total deliveries {total_deliveries} != 360"
assert all(0 <= r["economy"] <= 36 for r in results), "Economy out of bounds"
assert all(r["deliveries"] > 0 for r in results), "Zero deliveries for a bowler"
assert output_path.exists(), "Output file not written"
print(f"\nValidation passed: {total_deliveries} total deliveries across all bowlers")
print(f"Output written to: {output_path}")
print("\nEquivalent Hadoop Streaming command:")
print("hadoop jar hadoop-streaming.jar \\")
print(" -input hdfs:///ipl/deliveries/*.tsv \\")
print(" -output hdfs:///ipl/bowler_stats/ \\")
print(" -mapper mapper.py \\")
print(" -combiner combiner.py \\")
print(" -reducer reducer.py")