MapReduce Mechanics & Architecture

MapReduce Pipeline Lifecycle

graph TD
    A["Raw Data Blocks"] -->|"1. Input Split"| B["Input Splits"]
    B -->|"2. Map Phase"| C["Mappers (Emit Key-Values)"]
    C -->|"3. Shuffle & Sort"| D["Network Shuffle (Group by Key)"]
    D -->|"4. Reduce Phase"| E["Reducers (Aggregate per Key)"]
    E -->|"5. Output"| F["HDFS Final Files"]

    style C fill:#eff6ff,stroke:#2563eb,stroke-width:2px;
    style D fill:#fef2f2,stroke:#dc2626,stroke-width:2px;
    style E fill:#f0fdf4,stroke:#16a34a,stroke-width:2px;

Execution Stages:

1. Input Split & Record Reader: Logically divides data into 128MB splits and parses raw bytes into (key, value) pairs.
2. Map Phase: Emits intermediate pairs Map(k1, v1) → list(k2, v2) into an in-memory buffer (100MB), spilling to disk when 80% full.
3. Shuffle & Sort: Partitioner hashes keys Partition = hash(k2) % NumReducers and routes matching keys over TCP to the same Reducer.
4. Reduce Phase: Aggregates grouped value iterators Reduce(k2, list(v2)) → list(k3, v3) and writes final outputs directly to HDFS.

Engine Performance Optimizations

• Speculative Execution: If a worker node (straggler) runs slowly due to disk/hardware degradation, the Master launches a duplicate backup task on a healthy node, keeping whichever finishes first.
• The Combiner (Mini-Reducer): Runs locally on the mapper node to aggregate keys before network transfer (e.g. converting 5,000 ("spark", 1) tuples into a single ("spark", 5000) record), saving >90% network traffic.

Python MapReduce Engine Simulation

Below is a clean, runnable Python script simulating the Map, Shuffle, and Reduce stages locally:

from collections import defaultdict
import re

# 1. Input Blocks (Simulating HDFS data splits)
hdfs_blocks = ["Spark Hadoop", "Spark Pig", "Hive Hive"]

# 2. Mapper: Converts text into (key, value) pairs
def mapper(line):
    return [(word.lower(), 1) for word in re.findall(r'\b\w+\b', line)]

# 3. Reducer: Aggregates total count per word
def reducer(word, counts):
    return (word, sum(counts))

# Map Phase
mapper_outputs = []
for block in hdfs_blocks:
    mapper_outputs.extend(mapper(block))

# Shuffle & Sort Phase
shuffled_data = defaultdict(list)
for key, value in mapper_outputs:
    shuffled_data[key].append(value)

# Reduce Phase
final_results = [reducer(key, values) for key, values in sorted(shuffled_data.items())]

print("Final Word Counts:")
for word, count in final_results:
    print(f"  {word}: {count}")