🧠 MR Fundamentals

🎯 What is MapReduce?

MapReduce is a programming model and processing framework for distributed computing on large datasets. It splits work into independent map tasks that process key-value pairs, then reduces the results. It's the original Big Data processing paradigm, powering Hadoop and influencing Spark, Flink, and others.

Why MapReduce Matters

Scale: Processes petabytes. Simple: Just two functions. Parallel: Automatic distribution. Fault-tolerant: Handles failures. Proven: Google, Yahoo, Facebook. Foundation: Spark, Flink built on ideas.

📌 Map + Reduce📌 Key-Value Pairs📌 Distributed📌 Fault-Tolerant

# MapReduce — the big picture
# MapReduce is a programming model for distributed data processing.

# CONCEPTUAL OVERVIEW
print("""
=== MapReduce Overview ===

MapReduce processes data in two phases:

1. MAP
   - Input: (key, value) pairs
   - Output: list of (key, value) pairs
   - Runs in parallel across many machines
   
2. REDUCE
   - Input: (key, list of values)
   - Output: (key, value) pairs
   - Aggregates results from map phase

Example: Word Count
  Input:  "hello world hello"
  
  Map:
    (0, "hello world hello") → [(hello, 1), (world, 1), (hello, 1)]
  
  Shuffle & Sort:
    hello → [1, 1]
    world → [1]
  
  Reduce:
    (hello, [1, 1]) → (hello, 2)
    (world, [1])    → (world, 1)
  
  Output: hello: 2, world: 1
""")

# SIMPLE MAPREDUCE IMPLEMENTATION
from collections import defaultdict

def map_function(key, value):
    """
    Map function: emit (word, 1) for each word.
    """
    results = []
    for word in value.split():
        results.append((word, 1))
    return results

def reduce_function(key, values):
    """
    Reduce function: sum the values.
    """
    return (key, sum(values))

def mapreduce(data, mapper, reducer):
    """
    Simple MapReduce implementation.
    """
    # MAP PHASE
    mapped = []
    for key, value in data:
        mapped.extend(mapper(key, value))
    
    # SHUFFLE & SORT PHASE
    grouped = defaultdict(list)
    for key, value in mapped:
        grouped[key].append(value)
    
    # REDUCE PHASE
    results = []
    for key in sorted(grouped.keys()):
        results.append(reducer(key, grouped[key]))
    
    return results


# Test with word count
print("=== Word Count Example ===")
data = [
    (0, "hello world hello"),
    (1, "world of mapreduce"),
    (2, "hello mapreduce world"),
]

results = mapreduce(data, map_function, reduce_function)

print("\nWord counts:")
for word, count in results:
    print(f"  {word}: {count}")

# KEY PROPERTIES
print("""
=== Key Properties ===

1. DATA PARALLELISM
   - Same operation on many records
   - Automatic distribution
   - Scales with cluster size

2. KEY-VALUE MODEL
   - Everything is (key, value)
   - Flexible data representation
   - Easy to partition by key

3. FAULT TOLERANCE
   - Task retry on failure
   - Speculative execution
   - Data replication (HDFS)

4. DATA LOCALITY
   - Move computation to data
   - Minimize network traffic
   - HDFS blocks → map tasks

5. SIMPLICITY
   - Only implement map() and reduce()
   - Framework handles the rest
   - No distributed programming needed

6. SCALABILITY
   - Add more nodes
   - Linear scale
   - Petabytes of data
""")

# REAL-WORLD SCALE
print("=== Real-World MapReduce ===")
print("""
Google (2004 paper):
  - 1000s of machines
  - TB of data per job
  - Used for: indexing, analytics

Yahoo:
  - 40,000 nodes
  - 100,000+ jobs/day
  - Petabytes of data

Facebook:
  - 4+ PB/day
  - Thousands of jobs
  - Used for: analytics, ML

LinkedIn:
  - 3,000+ nodes
  - Used for: analytics, recommendations

Twitter:
  - 10,000+ nodes
  - Used for: analytics, trends
""")

📐 Programming Model

🎯 The MapReduce Model

MapReduce provides a simple programming model: you write two functions — map() and reduce(). The framework handles parallelization, distribution, fault tolerance, and data movement.

The Model

Map: (k1, v1) → list(k2, v2). Reduce: (k2, list(v2)) → list(k3, v3). Keys: Partition and sort. Values: Processed independently. Combiner: Local reduce. Partitioner: Assign to reducers.

📥 Map📤 Reduce🔑 Key-Value📊 Partitioner

# MapReduce Programming Model
print("=== Formal Definition ===")
print("""
MapReduce Model:

  MAP:     (k1, v1) → list(k2, v2)
  REDUCE:  (k2, list(v2)) → list(k3, v3)

Where:
  k1, v1 = input key-value pairs
  k2, v2 = intermediate key-value pairs
  k3, v3 = output key-value pairs

The framework:
  1. Splits input into chunks
  2. Runs map() on each chunk in parallel
  3. Shuffles and sorts intermediate results
  4. Groups values by key
  5. Runs reduce() on each key in parallel
  6. Writes output
""")

# MAP FUNCTION
print("\n=== Map Function ===")
print("""
map(key, value):
    # Process one input pair
    # Emit zero or more intermediate pairs
    for each something in value:
        emit(intermediate_key, intermediate_value)

Key points:
  - Input: one (key, value) pair
  - Output: list of (key, value) pairs
  - Stateless: no memory between calls
  - Runs in parallel

Example: Extract words
  map(0, "hello world"):
      emit("hello", 1)
      emit("world", 1)
""")

# REDUCE FUNCTION
print("\n=== Reduce Function ===")
print("""
reduce(key, values):
    # Process all values for one key
    # Emit zero or more output pairs
    result = aggregate(values)
    emit(key, result)

Key points:
  - Input: (key, list of values)
  - Output: list of (key, value) pairs
  - Values grouped by key
  - Runs in parallel by key

Example: Sum counts
  reduce("hello", [1, 1, 1]):
      emit("hello", 3)
""")

# COMBINER
print("\n=== Combiner (Optional) ===")
print("""
Combiner:
  - Runs after map, before reduce
  - Local aggregation
  - Reduces network traffic
  - Same interface as reducer

Example: Word Count
  Without combiner:
    Map output: (hello, 1), (hello, 1), (world, 1)
    Network: 3 pairs sent
  
  With combiner:
    Map output: (hello, 1), (hello, 1), (world, 1)
    Combiner: (hello, 2), (world, 1)
    Network: 2 pairs sent (less traffic!)

Requirements:
  - Commutative: a + b = b + a
  - Associative: (a + b) + c = a + (b + c)
""")

# PARTITIONER
print("\n=== Partitioner ===")
print("""
Partitioner:
  - Determines which reducer gets which key
  - Default: hash(key) % num_reducers
  - Ensures same key → same reducer

Example:
  hash("hello") % 3 = 1 → Reducer 1
  hash("world") % 3 = 2 → Reducer 2
  hash("hello") % 3 = 1 → Reducer 1

Custom partitioner:
  - For secondary sort
  - For load balancing
  - For range partitioning
""")

# COMPLETE EXAMPLE
print("\n=== Complete MapReduce Job ===")

def mapper(key, value):
    """Emit (word, 1) for each word."""
    for word in value.split():
        yield (word.lower(), 1)

def combiner(key, values):
    """Local aggregation."""
    yield (key, sum(values))

def reducer(key, values):
    """Sum all counts."""
    yield (key, sum(values))

def partitioner(key, num_reducers):
    """Default partitioner."""
    return hash(key) % num_reducers

def mapreduce_job(input_data, mapper, reducer, combiner=None, num_reducers=2):
    """Full MapReduce job."""
    # MAP
    mapped = []
    for key, value in input_data:
        mapped.extend(mapper(key, value))
    
    print(f"Map output: {mapped}")
    
    # COMBINE (optional)
    if combiner:
        combined = defaultdict(list)
        for key, value in mapped:
            combined[key].append(value)
        
        mapped = []
        for key, values in combined.items():
            mapped.extend(combiner(key, values))
        
        print(f"Combine output: {mapped}")
    
    # PARTITION
    partitioned = defaultdict(list)
    for key, value in mapped:
        p = partitioner(key, num_reducers)
        partitioned[p].append((key, value))
    
    print(f"Partitioned: {dict(partitioned)}")
    
    # SHUFFLE & SORT
    grouped = defaultdict(list)
    for key, value in mapped:
        grouped[key].append(value)
    
    # REDUCE
    results = []
    for key in sorted(grouped.keys()):
        results.extend(reducer(key, grouped[key]))
    
    return results


# Test
print("\n=== Running MapReduce Job ===")
data = [
    (0, "Hello World"),
    (1, "Hello MapReduce"),
    (2, "World of MapReduce"),
]

results = mapreduce_job(
    data,
    mapper,
    reducer,
    combiner=combiner,
    num_reducers=2
)

print(f"\nFinal output:")
for key, value in results:
    print(f"  {key}: {value}")

# DATA FLOW
print("\n=== Data Flow ===")
print("""
Input Splits → Map → Combine → Partition → Shuffle → Sort → Reduce → Output

1. INPUT SPLITS
   - HDFS blocks (128 MB)
   - One map task per split

2. MAP
   - Process each split
   - Emit intermediate pairs

3. COMBINE (optional)
   - Local aggregation
   - Reduce network traffic

4. PARTITION
   - Assign to reducers
   - hash(key) % num_reducers

5. SHUFFLE
   - Transfer to reducers
   - Network traffic

6. SORT
   - Sort by key
   - Group values

7. REDUCE
   - Process each key
   - Emit final output

8. OUTPUT
   - Write to HDFS
   - One file per reducer
""")

# MODEL SUMMARY
print("\n=== Model Summary ===")
print(f"{'Component':<20} {'Input':<25} {'Output':<25}")
print("-" * 70)
print(f"{'Map':<20} {'(k1, v1)':<25} {'list(k2, v2)':<25}")
print(f"{'Combine':<20} {'(k2, list(v2))':<25} {'list(k2, v2)':<25}")
print(f"{'Partition':<20} {'(k2, v2)':<25} {'reducer ID':<25}")
print(f"{'Shuffle':<20} {'All map outputs':<25} {'Grouped by key':<25}")
print(f"{'Sort':<20} {'Grouped by key':<25} {'Sorted by key':<25}")
print(f"{'Reduce':<20} {'(k2, list(v2))':<25} {'list(k3, v3)':<25}")

🔄 Map, Shuffle, Reduce

🎯 The Three Phases

MapReduce executes in three main phases: Map (parallel processing), Shuffle (data movement), and Reduce (aggregation). Understanding each phase is key to optimizing MapReduce jobs.

Phase Details

Map: Parallel, local. Shuffle: Network, expensive. Reduce: Parallel, aggregation. Sort: Part of shuffle. Combine: Map-side reduce. Bottleneck: Shuffle.

📥 Map🔀 Shuffle📤 Reduce📊 Sort

# MapReduce Phases
print("=== Phase 1: MAP ===")
print("""
Map Phase:
  - Input: Input splits (HDFS blocks)
  - Process: Apply map() to each record
  - Output: Intermediate (key, value) pairs
  - Parallelism: One map task per split
  - Data locality: Process local data

Characteristics:
  - Embarrassingly parallel
  - No communication between mappers
  - Stateless
  - Fast

Optimization:
  - Use combiner
  - Tune split size
  - Use appropriate data formats
""")

print("\n=== Phase 2: SHUFFLE ===")
print("""
Shuffle Phase:
  - Transfer map outputs to reducers
  - Sort by key
  - Group values by key
  - Network-intensive

Steps:
  1. Partition: hash(key) % num_reducers
  2. Sort: Sort by key within partition
  3. Spill: Write to disk if needed
  4. Merge: Merge sorted files
  5. Fetch: Reducers fetch from mappers

Bottleneck:
  - Network traffic
  - Disk I/O
  - Often the most expensive phase

Optimization:
  - Use combiner to reduce data
  - Compress map output
  - Tune buffer sizes
  - Increase reducers
""")

print("\n=== Phase 3: REDUCE ===")
print("""
Reduce Phase:
  - Input: (key, list of values)
  - Process: Apply reduce() to each key
  - Output: Final (key, value) pairs
  - Parallelism: One reduce task per partition

Characteristics:
  - Parallel by key
  - Values grouped
  - Sorted by key
  - Aggregation

Optimization:
  - Choose good partitioner
  - Tune number of reducers
  - Use secondary sort
""")

# DETAILED EXAMPLES
print("\n=== Example: Word Count Data Flow ===")
print("""
Input:
  "hello world hello"

Map:
  (hello, 1)
  (world, 1)
  (hello, 1)

Shuffle & Sort:
  hello → [1, 1]
  world → [1]

Reduce:
  (hello, 2)
  (world, 1)

Output:
  hello: 2
  world: 1
""")

# IMPLEMENTATION
print("\n=== Simulating Phases ===")

class MapReducePhases:
    """Simulate MapReduce phases in detail."""
    
    def __init__(self, num_mappers=2, num_reducers=2):
        self.num_mappers = num_mappers
        self.num_reducers = num_reducers
        self.map_outputs = []
        self.shuffled = {}
        self.reduced = []
    
    def map_phase(self, splits, mapper):
        """Phase 1: Map."""
        print(f"\n=== MAP PHASE ({len(splits)} splits) ===")
        
        for i, split in enumerate(splits):
            print(f"\nMapper {i}:")
            outputs = mapper(split)
            for key, value in outputs:
                print(f"  ({key}, {value})")
            self.map_outputs.extend(outputs)
    
    def shuffle_phase(self):
        """Phase 2: Shuffle and Sort."""
        print(f"\n=== SHUFFLE PHASE ===")
        
        # Partition
        partitions = defaultdict(list)
        for key, value in self.map_outputs:
            p = hash(key) % self.num_reducers
            partitions[p].append((key, value))
        
        print(f"\nPartitioned into {self.num_reducers} reducers:")
        for p, pairs in partitions.items():
            print(f"  Reducer {p}: {pairs}")
        
        # Sort and group
        print(f"\nSorted and grouped:")
        for p, pairs in partitions.items():
            grouped = defaultdict(list)
            for key, value in pairs:
                grouped[key].append(value)
            
            self.shuffled[p] = grouped
            print(f"  Reducer {p}:")
            for key in sorted(grouped.keys()):
                print(f"    {key} → {grouped[key]}")
    
    def reduce_phase(self, reducer):
        """Phase 3: Reduce."""
        print(f"\n=== REDUCE PHASE ===")
        
        for p in sorted(self.shuffled.keys()):
            print(f"\nReducer {p}:")
            for key in sorted(self.shuffled[p].keys()):
                values = self.shuffled[p][key]
                output = reducer(key, values)
                for k, v in output:
                    print(f"  ({k}, {v})")
                self.reduced.extend(output)
    
    def run(self, splits, mapper, reducer):
        self.map_phase(splits, mapper)
        self.shuffle_phase()
        self.reduce_phase(reducer)
        return self.reduced


# Example
def word_mapper(text):
    """Map: text → (word, 1) pairs."""
    return [(word.lower(), 1) for word in text.split()]

def word_reducer(key, values):
    """Reduce: sum counts."""
    return [(key, sum(values))]

# Run
mr = MapReducePhases(num_mappers=2, num_reducers=2)
splits = [
    "hello world hello",
    "world of mapreduce",
]

results = mr.run(splits, word_mapper, word_reducer)

print("\n=== FINAL OUTPUT ===")
for key, value in results:
    print(f"  {key}: {value}")

# PHASE COMPARISON
print("\n=== Phase Comparison ===")
print(f"{'Phase':<15} {'Input':<25} {'Output':<25} {'Cost':<15}")
print("-" * 80)
print(f"{'Map':<15} {'Input splits':<25} {'(k2, v2) pairs':<25} {'CPU':<15}")
print(f"{'Shuffle':<15} {'Map outputs':<25} {'Grouped by key':<25} {'Network':<15}")
print(f"{'Reduce':<15} {'(k2, [v2])':<25} {'(k3, v3)':<25} {'CPU':<15}")

# OPTIMIZATION TIPS
print("\n=== Optimization Tips ===")
print("""
1. MAP PHASE
   - Use combiner to reduce output
   - Read only needed columns (Parquet/ORC)
   - Increase split size for fewer mappers
   - Use data locality

2. SHUFFLE PHASE
   - Compress map output
   - Use combiner
   - Tune buffer sizes
   - Choose good partitioner
   - Avoid skew

3. REDUCE PHASE
   - Tune number of reducers
   - Use secondary sort
   - Avoid large values
   - Handle skew

4. GENERAL
   - Use appropriate data formats
   - Tune JVM settings
   - Monitor and profile
   - Use counters
""")

📝 Word Count Example

🎯 The "Hello World" of MapReduce

Word Count is the canonical MapReduce example: count the frequency of each word in a large text corpus. It demonstrates all phases and is the basis for many real applications.

Word Count Explained

Map: (doc, text) → (word, 1) for each word. Combine: Sum local counts. Shuffle: Group by word. Reduce: Sum all counts. Output: word → total count. Scale: Petabytes of text.

📝 Map🔢 Count📊 Sum🌍 Scale

# Word Count — the classic MapReduce example
print("=== Word Count Overview ===")
print("""
Problem: Count occurrences of each word in a large corpus.

Input: Text files (possibly TBs)
Output: (word, count) pairs

Map:    For each word, emit (word, 1)
Shuffle: Group by word
Reduce: Sum counts for each word

Example:
  Input: "the quick brown fox jumps over the lazy dog"
  
  Map:
    (the, 1), (quick, 1), (brown, 1), (fox, 1),
    (jumps, 1), (over, 1), (the, 1), (lazy, 1), (dog, 1)
  
  Shuffle:
    the → [1, 1]
    quick → [1]
    brown → [1]
    fox → [1]
    jumps → [1]
    over → [1]
    lazy → [1]
    dog → [1]
  
  Reduce:
    (the, 2), (quick, 1), (brown, 1), (fox, 1),
    (jumps, 1), (over, 1), (lazy, 1), (dog, 1)
""")

# COMPLETE IMPLEMENTATION
from collections import defaultdict
import re

class WordCount:
    """Complete Word Count implementation."""
    
    def __init__(self):
        self.map_outputs = []
        self.reduced = {}
    
    def mapper(self, key, value):
        """
        Map function: emit (word, 1) for each word.
        """
        # Tokenize: extract words
        words = re.findall(r'\b\w+\b', value.lower())
        
        for word in words:
            yield (word, 1)
    
    def combiner(self, key, values):
        """
        Combiner: local aggregation.
        """
        yield (key, sum(values))
    
    def reducer(self, key, values):
        """
        Reduce function: sum all counts.
        """
        yield (key, sum(values))
    
    def run(self, input_data, use_combiner=True):
        """
        Run word count.
        
        Args:
            input_data: list of (key, text) pairs
            use_combiner: whether to use combiner
        """
        print("=== MAP PHASE ===")
        mapped = []
        for key, value in input_data:
            for pair in self.mapper(key, value):
                mapped.append(pair)
                print(f"  Map: {key} → {pair}")
        
        self.map_outputs = mapped
        print(f"Map output: {len(mapped)} pairs")
        
        if use_combiner:
            print("\n=== COMBINE PHASE ===")
            combined = defaultdict(list)
            for key, value in mapped:
                combined[key].append(value)
            
            combined_output = []
            for key, values in combined.items():
                for pair in self.combiner(key, values):
                    combined_output.append(pair)
                    print(f"  Combine: {key} → {pair}")
            
            mapped = combined_output
            print(f"Combine output: {len(mapped)} pairs")
        
        print("\n=== SHUFFLE & SORT PHASE ===")
        grouped = defaultdict(list)
        for key, value in mapped:
            grouped[key].append(value)
        
        for key in sorted(grouped.keys()):
            print(f"  {key} → {grouped[key]}")
        
        print("\n=== REDUCE PHASE ===")
        results = {}
        for key in sorted(grouped.keys()):
            for k, v in self.reducer(key, grouped[key]):
                results[k] = v
                print(f"  Reduce: ({k}, {v})")
        
        self.reduced = results
        return results


# Example 1: Simple text
print("=== Example 1: Simple Text ===")

wc = WordCount()
input_data = [
    (0, "the quick brown fox"),
    (1, "jumps over the lazy dog"),
    (2, "the dog sleeps"),
]

results = wc.run(input_data, use_combiner=True)

print("\nFinal word counts:")
for word, count in sorted(results.items(), key=lambda x: -x[1]):
    print(f"  {word}: {count}")

# Example 2: Larger corpus
print("\n=== Example 2: Larger Corpus ===")

corpus = [
    (0, "Data is the new oil. Data drives decisions."),
    (1, "Big data requires big data tools. Data everywhere."),
    (2, "Machine learning needs data. Data is essential."),
    (3, "Data science is the future. Data data data!"),
]

wc2 = WordCount()
results2 = wc2.run(corpus, use_combiner=True)

print("\nTop words:")
sorted_words = sorted(results2.items(), key=lambda x: -x[1])
for word, count in sorted_words[:10]:
    print(f"  {word}: {count}")

# HADOOP WORD COUNT (Java)
print("\n=== Hadoop Word Count (Java) ===")
print("""
// Mapper
public class WordCountMapper 
    extends Mapper {
    
    private final static IntWritable one = new IntWritable(1);
    private Text word = new Text();
    
    public void map(LongWritable key, Text value, Context context)
        throws IOException, InterruptedException {
        
        String line = value.toString();
        StringTokenizer tokenizer = new StringTokenizer(line);
        
        while (tokenizer.hasMoreTokens()) {
            word.set(tokenizer.nextToken());
            context.write(word, one);
        }
    }
}

// Reducer
public class WordCountReducer 
    extends Reducer {
    
    public void reduce(Text key, Iterable values, 
                       Context context)
        throws IOException, InterruptedException {
        
        int sum = 0;
        for (IntWritable val : values) {
            sum += val.get();
        }
        context.write(key, new IntWritable(sum));
    }
}

// Driver
public class WordCount {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "word count");
        
        job.setJarByClass(WordCount.class);
        job.setMapperClass(WordCountMapper.class);
        job.setCombinerClass(WordCountReducer.class);
        job.setReducerClass(WordCountReducer.class);
        
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}
""")

# HADOOP STREAMING (Python)
print("\n=== Hadoop Streaming (Python) ===")
print("""
# Mapper (mapper.py)
#!/usr/bin/env python3
import sys

for line in sys.stdin:
    line = line.strip()
    words = line.split()
    for word in words:
        print(f'{word.lower()}\\t1')

# Reducer (reducer.py)
#!/usr/bin/env python3
import sys

current_word = None
current_count = 0

for line in sys.stdin:
    line = line.strip()
    word, count = line.split('\\t', 1)
    count = int(count)
    
    if current_word == word:
        current_count += count
    else:
        if current_word:
            print(f'{current_word}\\t{current_count}')
        current_word = word
        current_count = count

if current_word:
    print(f'{current_word}\\t{current_count}')

# Run with Hadoop Streaming
hadoop jar hadoop-streaming.jar \\
  -input /data/input \\
  -output /data/output \\
  -mapper mapper.py \\
  -reducer reducer.py \\
  -file mapper.py \\
  -file reducer.py
""")

# VARIATIONS
print("\n=== Word Count Variations ===")
print("""
1. CASE INSENSITIVE
   - Lowercase all words
   - "The" = "the"

2. STOP WORD REMOVAL
   - Remove common words (the, a, an)
   - Focus on meaningful words

3. STEMMING
   - Reduce to root form
   - "running" → "run"

4. N-GRAMS
   - Count phrases
   - "the quick" → 1

5. TF-IDF
   - Term frequency × inverse document frequency
   - Important words

6. SENTIMENT
   - Count positive/negative words
   - Sentiment analysis
""")

# PERFORMANCE
print("\n=== Word Count Performance ===")
print("""
Factors:
  - Input size (TB)
  - Number of mappers
  - Number of reducers
  - Combiner usage
  - Compression

Example: 1 TB text
  - Mappers: ~8000 (128 MB splits)
  - Map output: ~10 TB (without combiner)
  - With combiner: ~1 TB
  - Reducers: 100-1000
  - Runtime: minutes to hours

Optimization:
  - Use combiner (10x reduction)
  - Compress map output
  - Tune reducers
  - Use SSDs for shuffle
""")

🏗️ Architecture

🎯 MapReduce Architecture

MapReduce architecture consists of a JobTracker (master) and TaskTrackers (workers). In Hadoop 2.x+, YARN replaces JobTracker with ResourceManager and ApplicationMaster.

Architecture Components

Client: Submits jobs. JobTracker: Schedules tasks (Hadoop 1.x). TaskTracker: Runs tasks (Hadoop 1.x). ResourceManager: YARN master. NodeManager: YARN worker. ApplicationMaster: Job coordinator.

🏗️ JobTracker🏗️ TaskTracker🏗️ YARN🏗️ ApplicationMaster

# MapReduce Architecture
print("=== Hadoop 1.x Architecture ===")
print("""
  ┌─────────────────────────────────────────┐
  │              Client                     │
  │         (Job Submission)                │
  └─────────────────┬───────────────────────┘
                    │
                    ▼
  ┌─────────────────────────────────────────┐
  │           JobTracker                    │
  │         (Master Node)                   │
  │                                         │
  │  - Job scheduling                       │
  │  - Task monitoring                      │
  │  - Resource management                  │
  │  - Fault tolerance                      │
  └─────────────────┬───────────────────────┘
                    │
        ┌───────────┼───────────┐
        │           │           │
        ▼           ▼           ▼
  ┌─────────┐ ┌─────────┐ ┌─────────┐
  │TaskTracker│ │TaskTracker│ │TaskTracker│
  │  (Node)  │ │  (Node)  │ │  (Node)  │
  │          │ │          │ │          │
  │ Map Task │ │ Map Task │ │ Map Task │
  │ Reduce Task│ │Reduce Task│ │Reduce Task│
  └─────────┘ └─────────┘ └─────────┘

Limitations:
  - JobTracker bottleneck (4000 nodes max)
  - Single point of failure
  - Only MapReduce
""")

print("\n=== Hadoop 2.x Architecture (YARN) ===")
print("""
  ┌─────────────────────────────────────────┐
  │              Client                     │
  └─────────────────┬───────────────────────┘
                    │
                    ▼
  ┌─────────────────────────────────────────┐
  │         ResourceManager                 │
  │         (Master Node)                   │
  │                                         │
  │  - Cluster resource management          │
  │  - Scheduling                           │
  └─────────────────┬───────────────────────┘
                    │
        ┌───────────┼───────────┐
        │           │           │
        ▼           ▼           ▼
  ┌─────────┐ ┌─────────┐ ┌─────────┐
  │NodeManager│ │NodeManager│ │NodeManager│
  │  (Node)  │ │  (Node)  │ │  (Node)  │
  │          │ │          │ │          │
  │ Container│ │ Container│ │ Container│
  │  (AppMaster)│ │(Map Task)│ │(Reduce)│
  └─────────┘ └─────────┘ └─────────┘

Components:
  - ResourceManager: Cluster resources
  - NodeManager: Per-node agent
  - ApplicationMaster: Per-job coordinator
  - Container: Resource allocation

Benefits:
  - Scalability (10,000+ nodes)
  - Multi-tenancy (MapReduce, Spark, etc.)
  - Better resource utilization
""")

# JOB LIFECYCLE
print("\n=== Job Lifecycle ===")
print("""
1. SUBMIT
   - Client submits job
   - ResourceManager allocates ApplicationMaster

2. INITIALIZE
   - ApplicationMaster starts
   - Requests containers for tasks

3. MAP TASKS
   - Containers allocated
   - Map tasks run
   - Data locality

4. SHUFFLE
   - Map outputs transferred
   - Sorted and grouped

5. REDUCE TASKS
   - Reduce tasks run
   - Output written to HDFS

6. COMPLETE
   - ApplicationMaster unregisters
   - Client notified

7. CLEANUP
   - Containers released
   - Resources freed
""")

# IMPLEMENTATION
print("\n=== Simulating MapReduce Architecture ===")

import queue
import threading
import time

class ResourceManager:
    """Simulate YARN ResourceManager."""
    
    def __init__(self, num_nodes=3):
        self.num_nodes = num_nodes
        self.nodes = {f"node{i}": {'containers': 4, 'used': 0} 
                      for i in range(num_nodes)}
        self.jobs = []
    
    def submit_job(self, job):
        """Submit a job."""
        self.jobs.append(job)
        print(f"ResourceManager: Job {job['id']} submitted")
        return self._allocate_am(job)
    
    def _allocate_am(self, job):
        """Allocate ApplicationMaster container."""
        for node_id, node in self.nodes.items():
            if node['used'] < node['containers']:
                node['used'] += 1
                print(f"  Allocated AM on {node_id}")
                return node_id
        return None
    
    def allocate_container(self, job_id, task_type):
        """Allocate a container for a task."""
        for node_id, node in self.nodes.items():
            if node['used'] < node['containers']:
                node['used'] += 1
                print(f"  Allocated {task_type} container on {node_id}")
                return node_id
        return None
    
    def release_container(self, node_id):
        """Release a container."""
        if node_id in self.nodes:
            self.nodes[node_id]['used'] -= 1
    
    def get_status(self):
        return {
            'nodes': self.num_nodes,
            'containers': sum(n['containers'] for n in self.nodes.values()),
            'used': sum(n['used'] for n in self.nodes.values()),
            'jobs': len(self.jobs),
        }


class ApplicationMaster:
    """Simulate ApplicationMaster."""
    
    def __init__(self, job_id, rm):
        self.job_id = job_id
        self.rm = rm
        self.map_tasks = []
        self.reduce_tasks = []
        self.status = 'initializing'
    
    def run(self, num_maps, num_reduces):
        """Run the job."""
        self.status = 'running'
        print(f"\nApplicationMaster {self.job_id}: Starting job")
        
        # Allocate map containers
        print(f"\n  Allocating {num_maps} map containers:")
        for i in range(num_maps):
            node = self.rm.allocate_container(self.job_id, f"map{i}")
            if node:
                self.map_tasks.append({'id': f"map{i}", 'node': node})
        
        # Simulate map execution
        print(f"\n  Running map tasks:")
        for task in self.map_tasks:
            print(f"    {task['id']} on {task['node']}: processing...")
            time.sleep(0.1)
        
        # Allocate reduce containers
        print(f"\n  Allocating {num_reduces} reduce containers:")
        for i in range(num_reduces):
            node = self.rm.allocate_container(self.job_id, f"reduce{i}")
            if node:
                self.reduce_tasks.append({'id': f"reduce{i}", 'node': node})
        
        # Simulate reduce execution
        print(f"\n  Running reduce tasks:")
        for task in self.reduce_tasks:
            print(f"    {task['id']} on {task['node']}: reducing...")
            time.sleep(0.1)
        
        # Release containers
        for task in self.map_tasks + self.reduce_tasks:
            self.rm.release_container(task['node'])
        
        self.status = 'completed'
        print(f"\nApplicationMaster {self.job_id}: Job completed")


# Simulate
rm = ResourceManager(num_nodes=3)

print("=== MapReduce Cluster Simulation ===")
print(f"ResourceManager: {rm.num_nodes} nodes")

# Submit job
job = {'id': 'job_001', 'name': 'wordcount'}
am_node = rm.submit_job(job)

# Run job
am = ApplicationMaster('job_001', rm)
am.run(num_maps=4, num_reduces=2)

# Status
status = rm.get_status()
print(f"\nCluster status:")
for key, val in status.items():
    print(f"  {key}: {val}")

# ARCHITECTURE SUMMARY
print("\n=== Architecture Summary ===")
print(f"{'Component':<25} {'Hadoop 1.x':<25} {'Hadoop 2.x (YARN)':<25}")
print("-" * 80)
print(f"{'Master':<25} {'JobTracker':<25} {'ResourceManager':<25}")
print(f"{'Worker':<25} {'TaskTracker':<25} {'NodeManager':<25}")
print(f"{'Job Coordinator':<25} {'JobTracker':<25} {'ApplicationMaster':<25}")
print(f"{'Resource Unit':<25} {'Slot':<25} {'Container':<25}")
print(f"{'Scalability':<25} {'4,000 nodes':<25} {'10,000+ nodes':<25}")
print(f"{'Multi-tenancy':<25} {'No':<25} {'Yes':<25}")

🧶 YARN

🎯 Yet Another Resource Negotiator

YARN is Hadoop's resource management layer, introduced in Hadoop 2.x. It separates resource management from processing, allowing multiple processing frameworks (MapReduce, Spark, Flink) to share a cluster.

YARN Components

ResourceManager: Cluster resources. NodeManager: Per-node agent. ApplicationMaster: Per-job coordinator. Container: Resource allocation. Scheduler: Allocates resources. Multi-tenancy: Multiple frameworks.

🧶 ResourceManager🧶 NodeManager🧶 ApplicationMaster🧶 Container

# YARN — Yet Another Resource Negotiator
print("=== YARN Overview ===")
print("""
YARN separates:
  - Resource management (ResourceManager)
  - Job scheduling (ApplicationMaster)
  - Task execution (NodeManager)

Benefits:
  - Multi-tenancy: Multiple frameworks
  - Scalability: 10,000+ nodes
  - Utilization: Better resource use
  - Flexibility: Any processing model

Frameworks on YARN:
  - MapReduce
  - Spark
  - Flink
  - Storm
  - Tez
  - Giraph
""")

print("\n=== YARN Architecture ===")
print("""
  ┌─────────────────────────────────────────┐
  │           ResourceManager               │
  │                                         │
  │  ┌─────────────┐  ┌─────────────────┐  │
  │  │ Scheduler   │  │ Applications    │  │
  │  │             │  │ Manager         │  │
  │  └─────────────┘  └─────────────────┘  │
  └─────────────────┬───────────────────────┘
                    │
        ┌───────────┼───────────┐
        │           │           │
        ▼           ▼           ▼
  ┌─────────┐ ┌─────────┐ ┌─────────┐
  │NodeManager│ │NodeManager│ │NodeManager│
  │         │ │         │ │         │
  │ ┌─────┐ │ │ ┌─────┐ │ │ ┌─────┐ │
  │ │Container│ │ │Container│ │ │Container│ │
  │ │(AppMaster)│ │(Map Task)│ │(Reduce)│ │
  │ └─────┘ │ │ └─────┘ │ │ └─────┘ │
  └─────────┘ └─────────┘ └─────────┘
""")

print("\n=== YARN Components ===")
print("""
1. RESOURCEMANAGER (RM)
   - Master daemon
   - Manages cluster resources
   - Schedules applications
   
   Components:
   - Scheduler: Allocates resources
   - ApplicationsManager: Manages jobs

2. NODEMANAGER (NM)
   - Per-node agent
   - Manages containers
   - Reports to RM
   - Monitors resource usage

3. APPLICATIONMASTER (AM)
   - Per-application
   - Negotiates resources
   - Coordinates tasks
   - Runs in container

4. CONTAINER
   - Resource allocation
   - CPU, memory
   - Runs a task
   - Managed by NM

5. SCHEDULER
   - FIFO: First In First Out
   - Capacity: Queues with capacity
   - Fair: Equal share
   - Dominant Resource Fairness (DRF)
""")

# SCHEDULERS
print("\n=== YARN Schedulers ===")
print("""
1. FIFO SCHEDULER
   - First In, First Out
   - Simple
   - Head-of-line blocking
   - Not for shared clusters

2. CAPACITY SCHEDULER
   - Multiple queues
   - Guaranteed capacity
   - Elastic
   - Good for multi-tenant

3. FAIR SCHEDULER
   - Equal share
   - Dynamic
   - Preemption
   - Good for interactive

4. DOMINANT RESOURCE FAIRNESS (DRF)
   - Multi-resource fairness
   - Balances CPU and memory
   - Used by Fair Scheduler
""")

# IMPLEMENTATION
print("\n=== Simulating YARN ===")

import heapq
from enum import Enum

class ContainerState(Enum):
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"

class Container:
    """YARN Container."""
    def __init__(self, container_id, memory, vcores):
        self.id = container_id
        self.memory = memory  # MB
        self.vcores = vcores
        self.state = ContainerState.PENDING
        self.task = None
    
    def __repr__(self):
        return f"Container({self.id}, {self.memory}MB, {self.vcores}vcores, {self.state.value})"


class NodeManager:
    """YARN NodeManager."""
    def __init__(self, node_id, total_memory, total_vcores):
        self.node_id = node_id
        self.total_memory = total_memory
        self.total_vcores = total_vcores
        self.available_memory = total_memory
        self.available_vcores = total_vcores
        self.containers = {}
    
    def allocate(self, container_id, memory, vcores):
        """Allocate a container."""
        if memory > self.available_memory or vcores > self.available_vcores:
            return None
        
        container = Container(container_id, memory, vcores)
        container.state = ContainerState.RUNNING
        
        self.containers[container_id] = container
        self.available_memory -= memory
        self.available_vcores -= vcores
        
        return container
    
    def release(self, container_id):
        """Release a container."""
        if container_id in self.containers:
            container = self.containers[container_id]
            self.available_memory += container.memory
            self.available_vcores += container.vcores
            container.state = ContainerState.COMPLETED
            del self.containers[container_id]
            return True
        return False
    
    def status(self):
        return {
            'node_id': self.node_id,
            'total_memory': self.total_memory,
            'available_memory': self.available_memory,
            'total_vcores': self.total_vcores,
            'available_vcores': self.available_vcores,
            'containers': len(self.containers),
        }


class ResourceManager:
    """YARN ResourceManager."""
    def __init__(self):
        self.nodes = {}
        self.applications = {}
        self.next_container_id = 1
    
    def register_node(self, node):
        self.nodes[node.node_id] = node
        print(f"ResourceManager: Node {node.node_id} registered")
    
    def submit_application(self, app_id, memory_per_task, vcores_per_task):
        """Submit an application."""
        print(f"\nResourceManager: Application {app_id} submitted")
        self.applications[app_id] = {
            'id': app_id,
            'memory_per_task': memory_per_task,
            'vcores_per_task': vcores_per_task,
            'containers': [],
        }
        return True
    
    def allocate_container(self, app_id):
        """Allocate a container for an application."""
        if app_id not in self.applications:
            return None
        
        app = self.applications[app_id]
        
        # Find node with available resources
        for node_id, node in self.nodes.items():
            if (node.available_memory >= app['memory_per_task'] and
                node.available_vcores >= app['vcores_per_task']):
                
                container_id = f"container_{self.next_container_id}"
                self.next_container_id += 1
                
                container = node.allocate(
                    container_id,
                    app['memory_per_task'],
                    app['vcores_per_task']
                )
                
                if container:
                    app['containers'].append(container_id)
                    return container
        
        return None
    
    def release_container(self, node_id, container_id):
        """Release a container."""
        if node_id in self.nodes:
            return self.nodes[node_id].release(container_id)
        return False
    
    def cluster_status(self):
        """Get cluster status."""
        total_memory = sum(n.total_memory for n in self.nodes.values())
        available_memory = sum(n.available_memory for n in self.nodes.values())
        total_vcores = sum(n.total_vcores for n in self.nodes.values())
        available_vcores = sum(n.available_vcores for n in self.nodes.values())
        
        return {
            'nodes': len(self.nodes),
            'total_memory_mb': total_memory,
            'available_memory_mb': available_memory,
            'used_memory_mb': total_memory - available_memory,
            'total_vcores': total_vcores,
            'available_vcores': available_vcores,
            'used_vcores': total_vcores - available_vcores,
            'applications': len(self.applications),
        }


# Simulate
rm = ResourceManager()

# Register nodes
for i in range(3):
    nm = NodeManager(f"node{i}", total_memory=8192, total_vcores=8)
    rm.register_node(nm)

print(f"\nCluster: {len(rm.nodes)} nodes")

# Submit application
rm.submit_application("app_001", memory_per_task=1024, vcores_per_task=1)

# Allocate containers
print("\nAllocating containers:")
for i in range(6):
    container = rm.allocate_container("app_001")
    if container:
        print(f"  Allocated: {container}")
    else:
        print(f"  No resources available for container {i+1}")

# Status
status = rm.cluster_status()
print(f"\nCluster status:")
for key, val in status.items():
    if 'memory' in key:
        print(f"  {key}: {val} MB")
    else:
        print(f"  {key}: {val}")

# Release some containers
print("\nReleasing containers:")
for container_id in list(rm.applications['app_001']['containers'])[:3]:
    for node_id, node in rm.nodes.items():
        if container_id in node.containers:
            rm.release_container(node_id, container_id)
            print(f"  Released {container_id} from {node_id}")
            break

# Final status
status = rm.cluster_status()
print(f"\nFinal cluster status:")
print(f"  Used memory: {status['used_memory_mb']} MB")
print(f"  Used vcores: {status['used_vcores']}")

# YARN SUMMARY
print("\n=== YARN Summary ===")
print(f"{'Component':<25} {'Role':<40}")
print("-" * 65)
print(f"{'ResourceManager':<25} {'Cluster resource management':<40}")
print(f"{'NodeManager':<25} {'Per-node agent':<40}")
print(f"{'ApplicationMaster':<25} {'Per-job coordinator':<40}")
print(f"{'Container':<25} {'Resource allocation':<40}")
print(f"{'Scheduler':<25} {'Resource scheduling':<40}")
print(f"{'Client':<25} {'Submit jobs, monitor':<40}")

🛡️ Fault Tolerance

🎯 Handling Failures in MapReduce

MapReduce is designed for fault tolerance: task failures, node failures, and network issues are handled automatically through task retry, speculative execution, and data replication.

Fault Tolerance Mechanisms

Task retry: Re-run failed tasks. Speculative: Backup slow tasks. Data replication: HDFS 3x. Heartbeat: Detect failures. Checkpointing: Save progress. Master recovery: Restart JobTracker.

🛡️ Task Retry🛡️ Speculative🛡️ Replication🛡️ Heartbeat

# Fault Tolerance in MapReduce
print("=== Types of Failures ===")
print("""
1. TASK FAILURE
   - Task throws exception
   - Task crashes
   - Task hangs
   
   Solution: Retry task (default 4 attempts)

2. NODE FAILURE
   - Node crashes
   - Node becomes unresponsive
   - Disk failure
   
   Solution: Re-run tasks on other nodes

3. NETWORK FAILURE
   - Network partition
   - Packet loss
   - High latency
   
   Solution: Timeouts, retries

4. MASTER FAILURE
   - JobTracker crashes
   - ResourceManager crashes
   
   Solution: HA, restart

5. DATA CORRUPTION
   - Disk failure
   - Bit rot
   
   Solution: Checksums, HDFS replication
""")

print("\n=== Task Retry ===")
print("""
Default behavior:
  - Map task: 4 attempts
  - Reduce task: 4 attempts
  - Configurable: mapreduce.map.maxattempts

Retry process:
  1. Task fails
  2. JobTracker marks as failed
  3. Task rescheduled on different node
  4. Re-run from beginning

If all retries fail:
  - Job fails
  - User notified

Config:
  mapreduce.map.maxattempts = 4
  mapreduce.reduce.maxattempts = 4
""")

print("\n=== Speculative Execution ===")
print("""
Problem: Stragglers (slow tasks) delay job completion.

Solution: Speculative execution
  - Run backup copies of slow tasks
  - First to finish wins
  - Others killed

How it works:
  1. JobTracker monitors task progress
  2. Identifies slow tasks (stragglers)
  3. Launches speculative copies
  4. First completion wins
  5. Other copies killed

Benefits:
  - Reduces tail latency
  - Handles slow nodes
  - Improves job completion time

Config:
  mapreduce.map.speculative = true
  mapreduce.reduce.speculative = true
""")

print("\n=== Node Failure ===")
print("""
Detection:
  - Heartbeat timeout (default 10 min)
  - Node marked as dead

Recovery:
  1. Tasks on failed node re-run
  2. Map outputs lost → re-run maps
  3. Reduce outputs lost → re-run reduces
  4. Job continues

Data:
  - HDFS replicas (3x)
  - No data loss
  - Replicate under-replicated blocks

Config:
  mapreduce.task.timeout = 600000 (10 min)
""")

print("\n=== Master Failure ===")
print("""
JobTracker failure (Hadoop 1.x):
  - Single point of failure
  - Job lost
  - Manual restart

ResourceManager failure (YARN):
  - HA configuration
  - Active/Standby
  - Automatic failover
  - Jobs continue

ApplicationMaster failure:
  - Per-job
  - Restarted by ResourceManager
  - Job continues

Config:
  yarn.resourcemanager.ha.enabled = true
""")

# IMPLEMENTATION
print("\n=== Simulating Fault Tolerance ===")

import random
import time

class Task:
    """Simulate a MapReduce task."""
    
    def __init__(self, task_id, task_type):
        self.id = task_id
        self.type = task_type  # 'map' or 'reduce'
        self.attempts = 0
        self.max_attempts = 4
        self.status = 'pending'  # pending, running, success, failed
        self.start_time = None
        self.end_time = None
        self.node = None
    
    def run(self, failure_rate=0.2):
        """Run the task with possible failure."""
        self.attempts += 1
        self.status = 'running'
        self.start_time = time.time()
        
        # Simulate execution time
        duration = random.uniform(0.1, 0.5)
        time.sleep(duration)
        
        # Simulate failure
        if random.random() < failure_rate:
            self.status = 'failed'
            self.end_time = time.time()
            return False
        
        self.status = 'success'
        self.end_time = time.time()
        return True
    
    def __repr__(self):
        return f"Task({self.id}, {self.type}, {self.status}, attempts={self.attempts})"


class FaultTolerantJob:
    """Simulate a fault-tolerant MapReduce job."""
    
    def __init__(self, job_id, num_maps, num_reduces):
        self.job_id = job_id
        self.num_maps = num_maps
        self.num_reduces = num_reduces
        self.tasks = []
        self.completed = []
        self.failed = []
        
        # Create tasks
        for i in range(num_maps):
            self.tasks.append(Task(f"map_{i}", 'map'))
        for i in range(num_reduces):
            self.tasks.append(Task(f"reduce_{i}", 'reduce'))
    
    def run(self, failure_rate=0.2):
        """Run job with fault tolerance."""
        print(f"Job {self.job_id}: Starting")
        print(f"  Tasks: {len(self.tasks)} ({self.num_maps} map, {self.num_reduces} reduce)")
        
        for task in self.tasks:
            success = False
            
            while task.attempts < task.max_attempts and not success:
                print(f"\n  Running {task.id} (attempt {task.attempts + 1}):")
                success = task.run(failure_rate)
                
                if success:
                    print(f"    ✓ Success")
                    self.completed.append(task)
                else:
                    print(f"    ✗ Failed")
                    if task.attempts < task.max_attempts:
                        print(f"    Retrying...")
                        time.sleep(0.1)
            
            if not success:
                print(f"    ✗✗ Task {task.id} failed after {task.max_attempts} attempts")
                self.failed.append(task)
        
        # Job status
        print(f"\nJob {self.job_id} completed:")
        print(f"  Successful: {len(self.completed)}")
        print(f"  Failed: {len(self.failed)}")
        print(f"  Status: {'SUCCESS' if not self.failed else 'FAILED'}")
        
        return not self.failed


# Run simulation
random.seed(42)

job = FaultTolerantJob("job_001", num_maps=3, num_reduces=2)
success = job.run(failure_rate=0.3)

# SPECULATIVE EXECUTION
print("\n=== Speculative Execution Simulation ===")

class SpeculativeTask(Task):
    """Task with speculative execution."""
    
    def run(self, failure_rate=0.2, speed=1.0):
        """Run with variable speed."""
        self.attempts += 1
        self.status = 'running'
        self.start_time = time.time()
        
        # Simulate variable execution time
        duration = random.uniform(0.1, 0.5) / speed
        time.sleep(duration)
        
        if random.random() < failure_rate:
            self.status = 'failed'
            self.end_time = time.time()
            return False
        
        self.status = 'success'
        self.end_time = time.time()
        return True


class SpeculativeJob:
    """Job with speculative execution."""
    
    def __init__(self, job_id, num_tasks):
        self.job_id = job_id
        self.num_tasks = num_tasks
        self.tasks = []
    
    def run(self):
        """Run with speculative execution."""
        print(f"Job {self.job_id}: Running {self.num_tasks} tasks")
        print("Monitoring for stragglers...")
        
        # Simulate tasks with varying speeds
        for i in range(self.num_tasks):
            speed = random.uniform(0.5, 2.0)
            task = SpeculativeTask(f"task_{i}", 'map')
            
            print(f"\n  Task {i}: speed={speed:.1f}x")
            
            # Run original
            start = time.time()
            task.run(speed=speed)
            original_time = time.time() - start
            
            print(f"    Original: {original_time:.2f}s")
            
            # If slow (straggler), run speculative copy
            if original_time > 0.3:
                print(f"    Straggler detected! Running speculative copy...")
                spec_task = SpeculativeTask(f"task_{i}_spec", 'map')
                start = time.time()
                spec_task.run(speed=speed * 1.5)  # Faster copy
                spec_time = time.time() - start
                print(f"    Speculative: {spec_time:.2f}s")
                print(f"    Winner: {'original' if original_time < spec_time else 'speculative'}")


# Run speculative simulation
random.seed(123)
spec_job = SpeculativeJob("job_002", num_tasks=3)
spec_job.run()

# FAULT TOLERANCE SUMMARY
print("\n=== Fault Tolerance Summary ===")
print(f"{'Failure':<25} {'Detection':<25} {'Recovery':<30}")
print("-" * 80)
print(f"{'Task failure':<25} {'Exception':<25} {'Retry (4x)':<30}")
print(f"{'Node failure':<25} {'Heartbeat':<25} {'Re-run tasks':<30}")
print(f"{'Network failure':<25} {'Timeout':<25} {'Retry':<30}")
print(f"{'Master failure':<25} {'HA':<25} {'Failover':<30}")
print(f"{'Data corruption':<25} {'Checksum':<25} {'Replica':<30}")
print(f"{'Stragglers':<25} {'Progress':<25} {'Speculative':<30}")

⚡ Optimization

🎯 Making MapReduce Fast

MapReduce optimization focuses on reducing shuffle data, balancing load, and tuning parameters. The shuffle phase is often the bottleneck, so most optimizations target it.

Optimization Areas

Combiner: Reduce shuffle data. Compression: Less network. Partitioner: Balance load. Data formats: Columnar. Memory: Tune buffers. Parallelism: More reducers.

⚡ Combiner⚡ Compression⚡ Partitioner⚡ Data Formats

# MapReduce Optimization
print("=== 1. Combiner ===")
print("""
Problem: Map output can be huge
  - Mapper: 1 TB input → 10 TB intermediate
  - Shuffle: 10 TB over network
  - Bottleneck!

Solution: Combiner (map-side reduce)
  - Aggregate locally before shuffle
  - Same interface as reducer
  - 10x reduction possible

Example: Word Count
  Without combiner:
    Mapper emits: (hello, 1), (hello, 1), (hello, 1)
    Shuffle: 3 pairs
  
  With combiner:
    Combiner: (hello, 3)
    Shuffle: 1 pair

Requirements:
  - Commutative: a + b = b + a
  - Associative: (a + b) + c = a + (b + c)
""")

print("\n=== 2. Compression ===")
print("""
Compress map output and final output:

Map output compression:
  - Reduces shuffle data
  - Faster network transfer
  - CPU overhead
  - Config: mapreduce.map.output.compress

Codecs:
  - Snappy: Fast, moderate compression (recommended)
  - LZO: Fast, good compression
  - Gzip: Slower, better compression
  - Bzip2: Slowest, best compression

Final output compression:
  - Reduces HDFS storage
  - Config: mapreduce.output.fileoutputformat.compress
""")

print("\n=== 3. Partitioner ===")
print("""
Default: hash(key) % num_reducers

Problems:
  - Skew: Some keys much larger
  - Imbalance: Some reducers overloaded

Solutions:
  - Custom partitioner
  - Range partitioning
  - Salting (add random prefix)

Example: Skewed keys
  Key "USA" → 80% of data
  Key "UK" → 10%
  Key "France" → 10%
  
  Solution: Split "USA" into USA_1, USA_2, ...
""")

print("\n=== 4. Data Formats ===")
print("""
Text (CSV, JSON):
  - Human-readable
  - Large
  - Slow to parse

Columnar (Parquet, ORC):
  - Compressed
  - Fast to read
  - Column pruning
  - Predicate pushdown

Avro:
  - Row-based
  - Schema evolution
  - Good for write-heavy

Recommendations:
  - Input: Parquet or ORC
  - Intermediate: Avro or Sequence
  - Output: Parquet or ORC
""")

print("\n=== 5. Memory Tuning ===")
print("""
Key parameters:

Map:
  mapreduce.map.memory.mb = 1024
  mapreduce.map.java.opts = -Xmx820m
  mapreduce.task.io.sort.mb = 256
  mapreduce.map.sort.spill.percent = 0.8

Reduce:
  mapreduce.reduce.memory.mb = 2048
  mapreduce.reduce.java.opts = -Xmx1640m
  mapreduce.reduce.shuffle.parallelcopies = 10
  mapreduce.reduce.shuffle.input.buffer.percent = 0.7

General:
  mapreduce.task.timeout = 600000
  mapreduce.map.speculative = true
""")

print("\n=== 6. Number of Reducers ===")
print("""
Rules of thumb:
  - 0.95 × num_nodes × num_cores (max)
  - 1.75 × num_nodes × num_cores (best)
  
Example: 10 nodes, 8 cores
  - Min: 0.95 × 10 × 8 = 76
  - Best: 1.75 × 10 × 8 = 140

Considerations:
  - Too few: Underutilization
  - Too many: Overhead
  - Depends on data size

Config:
  mapreduce.job.reduces = 100
""")

print("\n=== 7. Data Locality ===")
print("""
Best: Data local (same node)
  - No network transfer
  - Fastest

Good: Same rack
  - Rack-local
  - Moderate speed

Worst: Different rack
  - Network transfer
  - Slowest

Optimization:
  - Increase split size
  - Use HDFS block locality
  - Delay scheduling
""")

# IMPLEMENTATION
print("\n=== Simulating Optimizations ===")

import time

class OptimizedMapReduce:
    """Compare optimized vs non-optimized MapReduce."""
    
    def __init__(self, data_size_mb, num_mappers=10, num_reducers=5):
        self.data_size_mb = data_size_mb
        self.num_mappers = num_mappers
        self.num_reducers = num_reducers
    
    def run_without_optimization(self):
        """Run without optimizations."""
        print("\n=== Without Optimization ===")
        
        # Map output = 10x input (typical)
        map_output_mb = self.data_size_mb * 10
        print(f"  Input: {self.data_size_mb} MB")
        print(f"  Map output: {map_output_mb} MB")
        
        # Shuffle time (100 MB/s)
        shuffle_time = map_output_mb / 100
        print(f"  Shuffle time: {shuffle_time:.1f} s")
        
        # Reduce time
        reduce_time = map_output_mb / 200
        print(f"  Reduce time: {reduce_time:.1f} s")
        
        total = shuffle_time + reduce_time
        print(f"  Total: {total:.1f} s")
        
        return total
    
    def run_with_combiner(self):
        """Run with combiner."""
        print("\n=== With Combiner ===")
        
        map_output_mb = self.data_size_mb * 10
        combined_mb = map_output_mb * 0.1  # 10x reduction
        print(f"  Input: {self.data_size_mb} MB")
        print(f"  Map output: {map_output_mb} MB")
        print(f"  After combiner: {combined_mb} MB")
        
        shuffle_time = combined_mb / 100
        print(f"  Shuffle time: {shuffle_time:.1f} s")
        
        reduce_time = combined_mb / 200
        print(f"  Reduce time: {reduce_time:.1f} s")
        
        total = shuffle_time + reduce_time
        print(f"  Total: {total:.1f} s")
        
        return total
    
    def run_with_compression(self):
        """Run with compression."""
        print("\n=== With Compression ===")
        
        map_output_mb = self.data_size_mb * 10
        compressed_mb = map_output_mb * 0.3  # 70% reduction
        print(f"  Input: {self.data_size_mb} MB")
        print(f"  Map output: {map_output_mb} MB")
        print(f"  After compression: {compressed_mb} MB")
        
        shuffle_time = compressed_mb / 100
        print(f"  Shuffle time: {shuffle_time:.1f} s")
        
        reduce_time = compressed_mb / 200
        print(f"  Reduce time: {reduce_time:.1f} s")
        
        total = shuffle_time + reduce_time
        print(f"  Total: {total:.1f} s")
        
        return total
    
    def run_with_both(self):
        """Run with combiner + compression."""
        print("\n=== With Combiner + Compression ===")
        
        map_output_mb = self.data_size_mb * 10
        combined_mb = map_output_mb * 0.1
        compressed_mb = combined_mb * 0.3
        print(f"  Input: {self.data_size_mb} MB")
        print(f"  Map output: {map_output_mb} MB")
        print(f"  After combiner: {combined_mb} MB")
        print(f"  After compression: {compressed_mb} MB")
        
        shuffle_time = compressed_mb / 100
        print(f"  Shuffle time: {shuffle_time:.1f} s")
        
        reduce_time = compressed_mb / 200
        print(f"  Reduce time: {reduce_time:.1f} s")
        
        total = shuffle_time + reduce_time
        print(f"  Total: {total:.1f} s")
        
        return total


# Compare
print("=== Optimization Comparison ===")
print("Data: 1000 MB input, 10x map expansion")

mr = OptimizedMapReduce(data_size_mb=1000)

t1 = mr.run_without_optimization()
t2 = mr.run_with_combiner()
t3 = mr.run_with_compression()
t4 = mr.run_with_both()

print(f"\n=== Speedup ===")
print(f"  Baseline: {t1:.1f} s")
print(f"  Combiner: {t2:.1f} s ({t1/t2:.1f}x faster)")
print(f"  Compression: {t3:.1f} s ({t1/t3:.1f}x faster)")
print(f"  Both: {t4:.1f} s ({t1/t4:.1f}x faster)")

# OPTIMIZATION CHECKLIST
print("\n=== Optimization Checklist ===")
print("""
✓ Use combiner when possible
✓ Compress map output
✓ Choose appropriate data formats (Parquet/ORC)
✓ Tune number of reducers
✓ Handle data skew
✓ Increase memory for shuffle
✓ Use data locality
✓ Avoid small files
✓ Use counters for monitoring
✓ Profile and iterate

Anti-patterns:
✗ Too many small files
✗ Skewed keys
✗ No combiner
✗ Uncompressed intermediate
✗ Too few/many reducers
""")

🎨 Design Patterns

🎯 Common MapReduce Patterns

MapReduce design patterns are reusable solutions to common data processing problems. They provide templates for structuring map and reduce functions.

Pattern Categories

Summarization: Aggregation. Filtering: Selection. Joins: Combining datasets. Organization: Sorting, partitioning. Iterative: Graph processing. Input/Output: Format conversion.

🎨 Summarization🎨 Filtering🎨 Joins🎨 Organization

# MapReduce Design Patterns
print("=== 1. SUMMARIZATION PATTERNS ===")
print("""
1.1 NUMERICAL SUMMARIZATION
   - Count, sum, average, min, max
   
   Example: Count words
     Map: (word, 1)
     Reduce: sum(values)

1.2 INVERTED INDEX
   - Word → list of documents
   
   Example: Search index
     Map: (word, doc_id)
     Reduce: (word, [doc_ids])

1.3 HISTOGRAM
   - Distribution of values
   
   Example: Age distribution
     Map: (age_bucket, 1)
     Reduce: sum(values)
""")

print("\n=== 2. FILTERING PATTERNS ===")
print("""
2.1 FILTERING
   - Keep only matching records
   
   Example: Filter errors
     Map: if "ERROR" in line: emit(line, 1)
     Reduce: identity

2.2 TOP N
   - Top N records by some metric
   
   Example: Top 10 products
     Map: (product, sales)
     Reduce: top 10 by sales

2.3 DISTINCT
   - Remove duplicates
   
   Example: Unique users
     Map: (user_id, None)
     Reduce: identity (one per key)

2.4 BLOOM FILTERING
   - Probabilistic membership
   - Efficient for large sets
""")

print("\n=== 3. JOIN PATTERNS ===")
print("""
3.1 REDUCE-SIDE JOIN
   - Join in reducer
   - Both datasets by same key
   
   Example: Users + Orders
     Map: (user_id, ("user", user_data))
          (user_id, ("order", order_data))
     Reduce: join by user_id

3.2 MAP-SIDE JOIN
   - One dataset fits in memory
   - Broadcast to all mappers
   
   Example: Large + Small
     Load small in memory
     Join in mapper

3.3 REPLICATED JOIN
   - Small dataset replicated
   - Join in mapper

3.4 COMPOSITE JOIN
   - Multiple keys
   - Custom partitioner
""")

print("\n=== 4. ORGANIZATION PATTERNS ===")
print("""
4.1 SORTING
   - Total order sort
   - Range partitioner
   
   Example: Sort by timestamp
     Map: (timestamp, record)
     Reduce: identity

4.2 PARTITIONING
   - Partition by key
   - Custom partitioner
   
   Example: Partition by country
     Partitioner: hash(country)

4.3 BINNING
   - Group into categories
   
   Example: Age groups
     Map: (age_group, record)

4.4 SHARDING
   - Distribute data
""")

print("\n=== 5. ITERATIVE PATTERNS ===")
print("""
5.1 GRAPH PROCESSING
   - Iterative MapReduce
   - PageRank, shortest path
   
   Example: PageRank
     Map: propagate rank
     Reduce: sum ranks

5.2 K-MEANS CLUSTERING
   - Iterative
   - Assign + update centroids

5.3 EXPECTATION MAXIMIZATION
   - Iterative
   - E-step + M-step
""")

# IMPLEMENTATION
print("\n=== Implementing Patterns ===")

from collections import defaultdict

# Pattern 1: Numerical Summarization
def numerical_summarization(data, mapper, reducer):
    """Generic summarization."""
    mapped = []
    for key, value in data:
        mapped.extend(mapper(key, value))
    
    grouped = defaultdict(list)
    for key, value in mapped:
        grouped[key].append(value)
    
    results = []
    for key in sorted(grouped.keys()):
        results.append(reducer(key, grouped[key]))
    
    return results

# Example: Count, sum, average
def sum_mapper(key, value):
    yield (value['category'], value['amount'])

def sum_reducer(key, values):
    return (key, sum(values))

def count_mapper(key, value):
    yield (value['category'], 1)

def count_reducer(key, values):
    return (key, len(values))

def avg_mapper(key, value):
    yield (value['category'], value['amount'])

def avg_reducer(key, values):
    return (key, sum(values) / len(values))


# Test
data = [
    (0, {'category': 'A', 'amount': 100}),
    (1, {'category': 'B', 'amount': 200}),
    (2, {'category': 'A', 'amount': 150}),
    (3, {'category': 'B', 'amount': 300}),
    (4, {'category': 'A', 'amount': 50}),
]

print("=== Summarization Example ===")
print("\nSum by category:")
sums = numerical_summarization(data, sum_mapper, sum_reducer)
for cat, total in sums:
    print(f"  {cat}: {total}")

print("\nCount by category:")
counts = numerical_summarization(data, count_mapper, count_reducer)
for cat, count in counts:
    print(f"  {cat}: {count}")

print("\nAverage by category:")
avgs = numerical_summarization(data, avg_mapper, avg_reducer)
for cat, avg in avgs:
    print(f"  {cat}: {avg:.1f}")

# Pattern 2: Inverted Index
print("\n=== Inverted Index Example ===")

def inverted_index_mapper(doc_id, text):
    """Map: emit (word, doc_id) for each word."""
    words = set(text.lower().split())
    for word in words:
        yield (word, doc_id)

def inverted_index_reducer(word, doc_ids):
    """Reduce: collect doc IDs."""
    return (word, sorted(doc_ids))

documents = [
    (1, "the quick brown fox"),
    (2, "the lazy dog"),
    (3, "quick brown dog"),
]

print("Inverted index:")
index = numerical_summarization(documents, inverted_index_mapper, inverted_index_reducer)
for word, docs in index:
    print(f"  {word}: {docs}")

# Pattern 3: Top N
print("\n=== Top N Example ===")

def top_n_mapper(key, value):
    yield (value['category'], value['amount'])

def top_n_reducer(key, values, n=2):
    """Get top N values."""
    top = sorted(values, reverse=True)[:n]
    return (key, top)

print("Top 2 amounts per category:")
top_results = numerical_summarization(data, top_n_mapper, 
                                       lambda k, v: top_n_reducer(k, v, n=2))
for cat, top in top_results:
    print(f"  {cat}: {top}")

# Pattern 4: Filtering
print("\n=== Filtering Example ===")

def filter_mapper(key, value):
    """Only emit if condition met."""
    if value['amount'] > 100:
        yield (value['category'], value['amount'])

def filter_reducer(key, values):
    return (key, values)

print("Amounts > 100:")
filtered = numerical_summarization(data, filter_mapper, filter_reducer)
for cat, amounts in filtered:
    print(f"  {cat}: {amounts}")

# Pattern 5: Distinct
print("\n=== Distinct Example ===")

def distinct_mapper(key, value):
    yield (value['category'], None)

def distinct_reducer(key, values):
    return (key, None)

print("Distinct categories:")
distinct = numerical_summarization(data, distinct_mapper, distinct_reducer)
for cat, _ in distinct:
    print(f"  {cat}")

# PATTERN SUMMARY
print("\n=== Pattern Summary ===")
print(f"{'Pattern':<25} {'Map Emits':<25} {'Reduce Computes':<25}")
print("-" * 75)
print(f"{'Sum':<25} {'(key, value)':<25} {'sum(values)':<25}")
print(f"{'Count':<25} {'(key, 1)':<25} {'sum(values)':<25}")
print(f"{'Average':<25} {'(key, value)':<25} {'sum/len':<25}")
print(f"{'Max':<25} {'(key, value)':<25} {'max(values)':<25}")
print(f"{'Min':<25} {'(key, value)':<25} {'min(values)':<25}")
print(f"{'Inverted Index':<25} {'(word, doc_id)':<25} {'list(doc_ids)':<25}")
print(f"{'Top N':<25} {'(key, value)':<25} {'top N':<25}")
print(f"{'Distinct':<25} {'(key, None)':<25} {'identity':<25}")
print(f"{'Filter':<25} {'(key, value)':<25} {'pass-through':<25}")

🌊 Hadoop Streaming

🎯 MapReduce in Any Language

Hadoop Streaming allows you to write MapReduce jobs in any language (Python, Ruby, Perl, etc.) using standard input/output. It's a powerful way to use non-Java languages with Hadoop.

Streaming Basics

Mapper: Reads stdin, writes stdout. Reducer: Reads stdin, writes stdout. Format: key\tvalue. Language: Any. Testing: Unix pipes. Debug: Local simulation.

🌊 stdin/stdout🌊 Any language🌊 key\tvalue🌊 Unix pipes

# Hadoop Streaming
print("=== What is Hadoop Streaming? ===")
print("""
Hadoop Streaming:
  - Write MapReduce in any language
  - Use stdin/stdout
  - No Java required
  - Perfect for scripting

Protocol:
  - Mapper: reads lines from stdin
  - Mapper: writes key\tvalue to stdout
  - Reducer: reads sorted key\tvalue from stdin
  - Reducer: writes key\tvalue to stdout

Key points:
  - Tab separates key and value
  - One record per line
  - Reducer receives sorted input
""")

# PYTHON MAPPER
print("\n=== Python Mapper ===")
print("""
#!/usr/bin/env python3
# mapper.py

import sys

def main():
    for line in sys.stdin:
        line = line.strip()
        words = line.split()
        for word in words:
            # Output: word\t1
            print(f'{word.lower()}\\t1')

if __name__ == '__main__':
    main()
""")

# PYTHON REDUCER
print("\n=== Python Reducer ===")
print("""
#!/usr/bin/env python3
# reducer.py

import sys

def main():
    current_word = None
    current_count = 0
    
    for line in sys.stdin:
        line = line.strip()
        word, count = line.split('\\t', 1)
        count = int(count)
        
        if current_word == word:
            current_count += count
        else:
            if current_word:
                print(f'{current_word}\\t{current_count}')
            current_word = word
            current_count = count
    
    if current_word:
        print(f'{current_word}\\t{current_count}')

if __name__ == '__main__':
    main()
""")

# TESTING
print("\n=== Testing Locally ===")
print("""
# Test with Unix pipes:
cat input.txt | python3 mapper.py | sort | python3 reducer.py

# Example:
echo "hello world hello" | python3 mapper.py
# Output:
# hello\t1
# world\t1
# hello\t1

echo "hello world hello" | python3 mapper.py | sort
# Output:
# hello\t1
# hello\t1
# world\t1

echo "hello world hello" | python3 mapper.py | sort | python3 reducer.py
# Output:
# hello\t2
# world\t1
""")

# RUNNING WITH HADOOP
print("\n=== Running with Hadoop ===")
print("""
# Basic command:
hadoop jar $HADOOP_HOME/share/hadoop/tools/lib/hadoop-streaming-*.jar \\
  -input /data/input \\
  -output /data/output \\
  -mapper mapper.py \\
  -reducer reducer.py \\
  -file mapper.py \\
  -file reducer.py

# With combiner:
hadoop jar hadoop-streaming.jar \\
  -input /data/input \\
  -output /data/output \\
  -mapper mapper.py \\
  -combiner reducer.py \\
  -reducer reducer.py \\
  -file mapper.py \\
  -file reducer.py

# With compression:
hadoop jar hadoop-streaming.jar \\
  -input /data/input \\
  -output /data/output \\
  -mapper mapper.py \\
  -reducer reducer.py \\
  -file mapper.py \\
  -file reducer.py \\
  -D mapreduce.map.output.compress=true \\
  -D mapreduce.map.output.compress.codec=org.apache.hadoop.io.compress.SnappyCodec
""")

# COMPLETE EXAMPLE
print("\n=== Complete Example: Log Analysis ===")
print("""
# mapper.py — extract status codes
#!/usr/bin/env python3
import sys
import re

for line in sys.stdin:
    line = line.strip()
    # Parse log line
    match = re.search(r'" (\d{3}) ', line)
    if match:
        status = match.group(1)
        print(f'{status}\\t1')

# reducer.py — count status codes
#!/usr/bin/env python3
import sys

current_status = None
current_count = 0

for line in sys.stdin:
    line = line.strip()
    status, count = line.split('\\t', 1)
    count = int(count)
    
    if current_status == status:
        current_count += count
    else:
        if current_status:
            print(f'{current_status}\\t{current_count}')
        current_status = status
        current_count = count

if current_status:
    print(f'{current_status}\\t{current_count}')
""")

# IMPLEMENTATION
print("\n=== Simulating Hadoop Streaming ===")

import subprocess
import os

# Create mapper and reducer files
mapper_code = '''#!/usr/bin/env python3
import sys

for line in sys.stdin:
    line = line.strip()
    for word in line.split():
        print(f"{word.lower()}\\t1")
'''

reducer_code = '''#!/usr/bin/env python3
import sys

current_word = None
current_count = 0

for line in sys.stdin:
    line = line.strip()
    if not line:
        continue
    word, count = line.split("\\t", 1)
    count = int(count)
    
    if current_word == word:
        current_count += count
    else:
        if current_word:
            print(f"{current_word}\\t{current_count}")
        current_word = word
        current_count = count

if current_word:
    print(f"{current_word}\\t{current_count}")
'''

# Write files
with open('/tmp/mapper.py', 'w') as f:
    f.write(mapper_code)
with open('/tmp/reducer.py', 'w') as f:
    f.write(reducer_code)

os.chmod('/tmp/mapper.py', 0o755)
os.chmod('/tmp/reducer.py', 0o755)

# Simulate streaming
def simulate_streaming(input_data):
    """Simulate Hadoop Streaming locally."""
    # Step 1: Run mapper
    mapper_input = '\n'.join(input_data)
    mapper_result = subprocess.run(
        ['python3', '/tmp/mapper.py'],
        input=mapper_input,
        capture_output=True,
        text=True
    )
    mapper_output = mapper_result.stdout.strip().split('\n')
    
    # Step 2: Sort (shuffle)
    sorted_output = sorted(mapper_output)
    
    # Step 3: Run reducer
    reducer_input = '\n'.join(sorted_output)
    reducer_result = subprocess.run(
        ['python3', '/tmp/reducer.py'],
        input=reducer_input,
        capture_output=True,
        text=True
    )
    reducer_output = reducer_result.stdout.strip()
    
    return mapper_output, sorted_output, reducer_output


# Test
print("=== Local Streaming Simulation ===")
input_data = [
    "hello world hello",
    "world of mapreduce",
    "hello mapreduce world",
]

mapper_out, sorted_out, reducer_out = simulate_streaming(input_data)

print("\nMapper output:")
for line in mapper_out:
    print(f"  {line}")

print("\nSorted (shuffle):")
for line in sorted_out:
    print(f"  {line}")

print("\nReducer output:")
for line in reducer_out.split('\n'):
    print(f"  {line}")

# ADVANTAGES
print("\n=== Advantages of Streaming ===")
print("""
1. LANGUAGE FLEXIBILITY
   - Python, Ruby, Perl, Bash
   - Any language with stdin/stdout
   - No Java required

2. RAPID PROTOTYPING
   - Quick to write
   - Test locally
   - Iterate fast

3. EXISTING CODE
   - Reuse scripts
   - Leverage libraries
   - Text processing

4. DEBUGGING
   - Test with Unix pipes
   - Local execution
   - Easy to debug

5. DATA SCIENCE
   - Python ecosystem
   - NumPy, Pandas
   - ML libraries
""")

# LIMITATIONS
print("\n=== Limitations of Streaming ===")
print("""
1. PERFORMANCE
   - Slower than Java
   - Process overhead
   - Serialization

2. TYPE SAFETY
   - Text only
   - No native types
   - Parsing overhead

3. COMPLEX ALGORITHMS
   - Hard for complex logic
   - No access to Hadoop APIs
   - Limited to key-value

4. DEBUGGING
   - Harder at scale
   - Limited logging
   - Error handling

5. DEPENDENCIES
   - Need to ship dependencies
   - -file option
   - Version conflicts
""")

# BEST PRACTICES
print("\n=== Best Practices ===")
print("""
1. TEST LOCALLY FIRST
   cat input | mapper | sort | reducer

2. HANDLE ERRORS
   try/except in Python
   Log to stderr

3. USE COMBINER
   Same as reducer when possible

4. OPTIMIZE I/O
   Batch writes
   Avoid print per record

5. SHIP DEPENDENCIES
   -file for scripts
   -archives for libraries

6. USE APPROPRIATE TYPES
   Parse ints/floats
   Handle missing values

7. MONITOR
   Use counters
   Check stderr logs
""")

# STREAMING SUMMARY
print("\n=== Streaming Summary ===")
print(f"{'Aspect':<25} {'Description':<40}")
print("-" * 65)
print(f"{'Languages':<25} {'Any (Python, Ruby, etc.)':<40}")
print(f"{'Interface':<25} {'stdin/stdout':<40}")
print(f"{'Format':<25} {'key\\tvalue':<40}")
print(f"{'Testing':<25} {'Unix pipes':<40}")
print(f"{'Performance':<25} {'Slower than Java':<40}")
print(f"{'Use case':<25} {'Prototyping, text processing':<40}")

⚖️ MR vs Spark

🎯 MapReduce vs Apache Spark

MapReduce and Spark are both distributed processing frameworks. Spark is faster (in-memory) and more flexible (batch, streaming, ML), but MapReduce is simpler and more mature for batch processing.

Comparison

MapReduce: Disk-based, batch, Java. Spark: In-memory, multi-purpose, Scala/Python. Speed: Spark 100x faster. Cost: Spark needs more RAM. Maturity: MR more mature. Ecosystem: Spark broader.

⚖️ MapReduce⚖️ Spark⚖️ Disk vs Memory⚖️ Batch vs Multi

# MapReduce vs Spark
print("=== Architecture Comparison ===")
print("""
MAPREDUCE:
  - Disk-based
  - Map → Shuffle → Reduce
  - Each job reads/writes HDFS
  - Batch only

SPARK:
  - Memory-based
  - DAG of transformations
  - Data cached in memory
  - Batch + streaming + ML
""")

print("\n=== Performance Comparison ===")
print(f"{'Aspect':<25} {'MapReduce':<25} {'Spark':<25}")
print("-" * 75)
print(f"{'Storage':<25} {'Disk':<25} {'Memory':<25}")
print(f"{'Speed':<25} {'Slow':<25} {'100x faster':<25}")
print(f"{'Iterative':<25} {'Slow':<25} {'Fast':<25}")
print(f"{'Latency':<25} {'High':<25} {'Low':<25}")
print(f"{'Throughput':<25} {'High':<25} {'High':<25}")
print(f"{'Fault tolerance':<25} {'Task retry':<25} {'RDD lineage':<25}")
print(f"{'Memory':<25} {'Low':<25} {'High':<25}")
print(f"{'Cost':<25} {'Lower':<25} {'Higher':<25}")

print("\n=== Programming Model ===")
print("""
MAPREDUCE:
  - Map + Reduce functions
  - Java API (or streaming)
  - Key-value pairs
  - Explicit phases

SPARK:
  - RDDs / DataFrames
  - Scala, Python, Java, R
  - Functional transformations
  - Lazy evaluation
  - DAG optimization

Example: Word Count

MapReduce (Java):
  map(key, value):
      for word in value.split():
          emit(word, 1)
  
  reduce(key, values):
      emit(key, sum(values))

Spark (Python):
  counts = sc.textFile("input") \\
             .flatMap(lambda line: line.split()) \\
             .map(lambda word: (word, 1)) \\
             .reduceByKey(lambda a, b: a + b)
""")

print("\n=== When to Use Which ===")
print("""
USE MAPREDUCE WHEN:
  ✓ Very large datasets that don't fit in memory
  ✓ Simple batch processing
  ✓ Cost-sensitive (less memory)
  ✓ Existing Java codebase
  ✓ One-time batch jobs

USE SPARK WHEN:
  ✓ Iterative algorithms (ML)
  ✓ Interactive analytics
  ✓ Real-time streaming
  ✓ Complex pipelines
  ✓ Data fits in memory

MODERN TREND:
  - Spark is more popular
  - MapReduce still used for batch
  - Spark can run on YARN
  - Many orgs use both
""")

# IMPLEMENTATION COMPARISON
print("\n=== Implementation Comparison ===")

# Simulate MapReduce word count
def mapreduce_wordcount(data):
    """MapReduce-style word count."""
    # Map
    mapped = []
    for line in data:
        for word in line.split():
            mapped.append((word.lower(), 1))
    
    # Shuffle & Sort
    grouped = defaultdict(list)
    for word, count in mapped:
        grouped[word].append(count)
    
    # Reduce
    results = {}
    for word in sorted(grouped.keys()):
        results[word] = sum(grouped[word])
    
    return results

# Simulate Spark word count (functionally similar)
def spark_wordcount(data):
    """Spark-style word count."""
    # RDD operations
    words = []
    for line in data:
        words.extend(line.lower().split())
    
    # map
    pairs = [(word, 1) for word in words]
    
    # reduceByKey
    grouped = defaultdict(list)
    for word, count in pairs:
        grouped[word].append(count)
    
    results = {word: sum(counts) for word, counts in grouped.items()}
    
    return results

# Test
data = [
    "hello world hello",
    "world of mapreduce",
    "hello spark world",
]

print("=== Word Count Comparison ===")
mr_result = mapreduce_wordcount(data)
spark_result = spark_wordcount(data)

print("\nMapReduce result:")
for word, count in sorted(mr_result.items()):
    print(f"  {word}: {count}")

print("\nSpark result:")
for word, count in sorted(spark_result.items()):
    print(f"  {word}: {count}")

print("\nResults are identical!")

# PERFORMANCE SIMULATION
print("\n=== Performance Simulation ===")

import time

def simulate_disk_io(data_size_mb, iterations=1):
    """Simulate disk-based processing."""
    time_per_mb = 0.01  # 10ms per MB
    return data_size_mb * time_per_mb * iterations

def simulate_memory_io(data_size_mb, iterations=1):
    """Simulate memory-based processing."""
    time_per_mb = 0.001  # 1ms per MB
    # First iteration loads to memory
    load_time = data_size_mb * 0.005  # 5ms per MB
    return load_time + data_size_mb * time_per_mb * iterations

data_sizes = [100, 1000, 10000]
print(f"{'Data (MB)':<15} {'MapReduce':<15} {'Spark':<15} {'Speedup':<15}")
print("-" * 60)

for size in data_sizes:
    mr_time = simulate_disk_io(size, iterations=10)
    spark_time = simulate_memory_io(size, iterations=10)
    speedup = mr_time / spark_time
    print(f"{size:<15} {mr_time:.2f}s{'':<8} {spark_time:.2f}s{'':<8} {speedup:.1f}x")

print("""
Note: Real speedups are typically 10-100x for iterative algorithms.
""")

# COMPARISON SUMMARY
print("\n=== Summary ===")
print(f"{'Feature':<25} {'MapReduce':<25} {'Spark':<25}")
print("-" * 75)
print(f"{'First release':<25} {'2004 (Google)':<25} {'2014':<25}")
print(f"{'Language':<25} {'Java':<25} {'Scala, Python, Java, R':<25}")
print(f"{'Storage':<25} {'Disk':<25} {'Memory':<25}")
print(f"{'Speed':<25} {'Baseline':<25} {'10-100x faster':<25}")
print(f"{'Processing':<25} {'Batch':<25} {'Batch + Stream + ML':<25}")
print(f"{'Fault tolerance':<25} {'Task retry':<25} {'RDD lineage':<25}")
print(f"{'API':<25} {'Low-level':<25} {'High-level':<25}")
print(f"{'Ecosystem':<25} {'Hadoop':<25} {'Spark ecosystem':<25}")

💼 Applications

🎯 MapReduce in the Real World

MapReduce powers many real-world applications: search indexing, log analysis, machine learning, bioinformatics, and more. It's the workhorse of Big Data processing.

Application Areas

Search: Google index. Social: Facebook analytics. Retail: Recommendations. Finance: Risk analysis. Science: Genomics. Logs: Monitoring.

🔍 Search📊 Analytics🧬 Science💰 Finance

# MapReduce Applications
print("=== 1. Search Indexing ===")
print("""
Google's original use case:
  - Crawl web pages
  - Build inverted index
  - PageRank computation

MapReduce jobs:
  - Index: (word, doc_id) pairs
  - PageRank: iterative
  - Link analysis

Example: Inverted Index
  Map: (doc_id, text) → [(word, doc_id) for word in text]
  Reduce: (word, [doc_ids]) → (word, list)

Scale: Billions of pages
""")

print("\n=== 2. Log Analysis ===")
print("""
Analyze server logs:
  - Error rates
  - Traffic patterns
  - User behavior

MapReduce jobs:
  - Count errors by type
  - Aggregate by hour/day
  - Detect anomalies

Example: Error Count
  Map: parse log, emit (error_type, 1)
  Reduce: sum counts

Scale: TBs of logs per day
""")

print("\n=== 3. Machine Learning ===")
print("""
Distributed ML:
  - Training on large datasets
  - Feature engineering
  - Model evaluation

MapReduce jobs:
  - K-means clustering
  - Naive Bayes
  - Linear regression
  - PageRank

Example: K-means
  Map: assign points to nearest centroid
  Reduce: recompute centroids
  Repeat until convergence
""")

print("\n=== 4. Bioinformatics ===")
print("""
Genomic analysis:
  - Sequence alignment
  - Variant calling
  - Gene expression

MapReduce jobs:
  - BLAST (sequence search)
  - Read mapping
  - Quality control

Example: Sequence Alignment
  Map: align each read to reference
  Reduce: aggregate results

Scale: TBs of genomic data
""")

print("\n=== 5. Finance ===")
print("""
Financial analytics:
  - Risk analysis
  - Fraud detection
  - Portfolio optimization

MapReduce jobs:
  - Monte Carlo simulation
  - VaR calculation
  - Transaction analysis

Example: Fraud Detection
  Map: analyze transaction
  Reduce: aggregate suspicious activity
""")

print("\n=== 6. E-commerce ===")
print("""
Retail analytics:
  - Recommendations
  - Customer segmentation
  - Demand forecasting

MapReduce jobs:
  - Collaborative filtering
  - Market basket analysis
  - Clickstream analysis

Example: Recommendations
  Map: (user, item) → (item, user)
  Reduce: find similar users
""")

print("\n=== 7. Social Media ===")
print("""
Social analytics:
  - Trend detection
  - Sentiment analysis
  - Community detection

MapReduce jobs:
  - Count hashtags
  - Sentiment scoring
  - Graph analysis

Example: Trending Topics
  Map: (hashtag, 1)
  Reduce: sum counts
  Sort by count
""")

# IMPLEMENTATION
print("\n=== Practical Example: Log Analyzer ===")

import re

class LogAnalyzer:
    """MapReduce for log analysis."""
    
    def __init__(self):
        self.map_outputs = []
    
    def mapper(self, line):
        """Parse log line, emit (status, 1)."""
        # Example log format:
        # 192.168.1.1 - - [01/Jan/2024:12:00:00] "GET /page HTTP/1.1" 200 1234
        
        match = re.search(r'" (\d{3}) ', line)
        if match:
            status = match.group(1)
            yield (status, 1)
    
    def reducer(self, key, values):
        """Count occurrences."""
        yield (key, sum(values))
    
    def analyze(self, logs):
        """Run MapReduce on logs."""
        # Map
        for line in logs:
            for pair in self.mapper(line):
                self.map_outputs.append(pair)
        
        # Shuffle & Sort
        grouped = defaultdict(list)
        for status, count in self.map_outputs:
            grouped[status].append(count)
        
        # Reduce
        results = {}
        for status in sorted(grouped.keys()):
            for key, count in self.reducer(status, grouped[status]):
                results[key] = count
        
        return results


# Test
logs = [
    '192.168.1.1 - - [01/Jan/2024:12:00:00] "GET /page HTTP/1.1" 200 1234',
    '192.168.1.2 - - [01/Jan/2024:12:00:01] "GET /error HTTP/1.1" 404 0',
    '192.168.1.1 - - [01/Jan/2024:12:00:02] "GET /page HTTP/1.1" 200 1234',
    '192.168.1.3 - - [01/Jan/2024:12:00:03] "POST /api HTTP/1.1" 500 0',
    '192.168.1.1 - - [01/Jan/2024:12:00:04] "GET /page HTTP/1.1" 200 1234',
    '192.168.1.2 - - [01/Jan/2024:12:00:05] "GET /missing HTTP/1.1" 404 0',
]

analyzer = LogAnalyzer()
results = analyzer.analyze(logs)

print("=== Log Analysis Results ===")
print("Status code counts:")
for status, count in sorted(results.items()):
    print(f"  {status}: {count}")

print("\n=== Error Analysis ===")
error_count = sum(count for status, count in results.items() if status.startswith('4') or status.startswith('5'))
print(f"  Total errors: {error_count}")
print(f"  Error rate: {error_count / sum(results.values()) * 100:.1f}%")

# APPLICATION SUMMARY
print("\n=== Application Summary ===")
print(f"{'Domain':<20} {'Use Case':<30} {'Scale':<20}")
print("-" * 70)
print(f"{'Search':<20} {'Indexing, PageRank':<30} {'Billions of pages':<20}")
print(f"{'Logs':<20} {'Error analysis':<30} {'TBs/day':<20}")
print(f"{'ML':<20} {'K-means, regression':<30} {'TBs':<20}")
print(f"{'Bio':<20} {'Sequence alignment':<30} {'TBs':<20}")
print(f"{'Finance':<20} {'Risk, fraud':<30} {'TBs':<20}")
print(f"{'Retail':<20} {'Recommendations':<30} {'TBs':<20}")
print(f"{'Social':<20} {'Trends, sentiment':<30} {'PBs':<20}")

⚠️ Limitations

🎯 Where MapReduce Falls Short

MapReduce has limitations: high latency, disk I/O, iterative algorithms, and real-time processing. Understanding these helps choose the right tool.

Key Limitations

Latency: High startup cost. Disk I/O: Every phase. Iterative: Inefficient. Real-time: Not suitable. Complex: Hard for some algorithms. Memory: Limited.

⚠️ Latency⚠️ Disk I/O⚠️ Iterative⚠️ Real-time

# MapReduce Limitations
print("=== 1. High Latency ===")
print("""
Problem:
  - Job startup overhead
  - Disk I/O between phases
  - Minutes to hours for small jobs

Why:
  - JVM startup
  - Task scheduling
  - Disk reads/writes

Impact:
  - Not suitable for interactive queries
  - Not suitable for real-time
  - Not for small datasets

Solution:
  - Use Spark for interactive
  - Use Hive/Impala for SQL
  - Use Kafka/Flink for streaming
""")

print("\n=== 2. Disk I/O ===")
print("""
Problem:
  - Every phase reads/writes HDFS
  - Map → disk → reduce → disk
  - Slow compared to memory

Example: Iterative algorithm
  Iteration 1: Read HDFS → Map → Reduce → Write HDFS
  Iteration 2: Read HDFS → Map → Reduce → Write HDFS
  ...
  
  Each iteration: full disk I/O

Solution:
  - Spark (in-memory)
  - Caching
  - Better algorithms
""")

print("\n=== 3. Iterative Algorithms ===")
print("""
Problem:
  - MapReduce is not designed for iteration
  - Each iteration is a separate job
  - Overhead per iteration

Examples:
  - K-means clustering
  - PageRank
  - Gradient descent
  - Graph algorithms

Impact:
  - Slow for ML
  - Slow for graph processing

Solution:
  - Spark (caches data in memory)
  - GraphX (graph processing)
  - Specialized frameworks
""")

print("\n=== 4. Real-Time Processing ===")
print("""
Problem:
  - MapReduce is batch-oriented
  - No streaming support
  - High latency

Impact:
  - Can't process real-time data
  - Can't do alerts
  - Can't do interactive

Solution:
  - Kafka (streaming)
  - Storm (real-time)
  - Flink (streaming)
  - Spark Streaming
""")

print("\n=== 5. Complex Algorithms ===")
print("""
Problem:
  - Some algorithms don't fit MapReduce
  - Graph algorithms
  - Recursive algorithms

Examples:
  - Shortest path
  - Connected components
  - PageRank (iterative)

Impact:
  - Hard to implement
  - Inefficient
  - May need multiple jobs

Solution:
  - GraphX (Spark)
  - Giraph (Hadoop)
  - Pregel (Google)
""")

print("\n=== 6. Memory Limitations ===")
print("""
Problem:
  - MapReduce uses disk
  - Limited memory
  - Can't cache

Impact:
  - Slow for iterative
  - Can't do in-memory joins
  - Limited optimization

Solution:
  - Spark (memory)
  - Increase memory
  - Better algorithms
""")

print("\n=== 7. Development Complexity ===")
print("""
Problem:
  - Java boilerplate
  - Complex APIs
  - Hard to debug

Impact:
  - Slow development
  - Hard to maintain
  - Steep learning curve

Solution:
  - Spark (simpler API)
  - Hive (SQL)
  - Pig (scripting)
  - Hadoop Streaming
""")

# COMPARISON
print("\n=== MapReduce vs Alternatives ===")
print(f"{'Requirement':<25} {'MapReduce':<20} {'Better Alternative':<25}")
print("-" * 70)
print(f"{'Interactive queries':<25} {'Slow':<20} {'Spark, Impala':<25}")
print(f"{'Iterative ML':<25} {'Slow':<20} {'Spark MLlib':<25}")
print(f"{'Real-time':<25} {'No':<20} {'Flink, Storm':<25}")
print(f"{'Graph processing':<25} {'Hard':<20} {'GraphX, Giraph':<25}")
print(f"{'SQL':<25} {'Hive':<20} {'Presto, Spark SQL':<25}")
print(f"{'Simple batch':<25} {'Good':<20} {'MapReduce':<25}")

# WHEN TO USE
print("\n=== When to Use MapReduce ===")
print("""
✓ Very large datasets (TB+)
✓ Simple batch processing
✓ One-time jobs
✓ Cost-sensitive
✓ Existing Java code
✓ When Spark is overkill

✗ Interactive queries
✗ Iterative algorithms
✗ Real-time processing
✗ Small datasets
✗ Graph algorithms
✗ Complex ML
""")

# LIMITATIONS SUMMARY
print("\n=== Limitations Summary ===")
print(f"{'Limitation':<25} {'Impact':<30} {'Workaround':<25}")
print("-" * 80)
print(f"{'High latency':<25} {'Slow for small jobs':<30} {'Spark':<25}")
print(f"{'Disk I/O':<25} {'Slow for iterative':<30} {'Spark':<25}")
print(f"{'Iterative':<25} {'Inefficient':<30} {'Spark, GraphX':<25}")
print(f"{'Real-time':<25} {'Not suitable':<30} {'Flink, Storm':<25}")
print(f"{'Complex algorithms':<25} {'Hard to implement':<30} {'Specialized':<25}")
print(f"{'Memory':<25} {'Limited caching':<30} {'Spark':<25}")
print(f"{'Development':<25} {'Java boilerplate':<30} {'Streaming, Hive':<25}")

🚀 Future: MR in Modern Computing

🎯 Where MapReduce Is Going

MapReduce is evolving: cloud integration, Spark dominance, and new architectures. While Spark has largely replaced MapReduce for new projects, MapReduce concepts remain foundational.

Future Directions

Cloud: Managed services. Spark: Dominant. Serverless: No infrastructure. Unified: Batch + stream. AI: ML pipelines. Legacy: Maintenance mode.

☁️ Cloud⚡ Spark🤖 AI🔧 Legacy

# Future of MapReduce
print("=== 1. Spark Dominance ===")
print("""
Trend: Spark replaces MapReduce

Reasons:
  - 100x faster (in-memory)
  - Unified API (batch + stream + ML)
  - Better developer experience
  - Active development

Current state:
  - New projects use Spark
  - MapReduce for legacy
  - Many orgs migrating

Future:
  - Spark as default
  - MapReduce for maintenance
  - Concepts remain important
""")

print("\n=== 2. Cloud Integration ===")
print("""
Trend: Managed services

Cloud options:
  - AWS EMR (MapReduce + Spark)
  - Google Dataproc
  - Azure HDInsight

Benefits:
  - No infrastructure
  - Auto-scaling
  - Pay per use

Serverless:
  - AWS Lambda
  - Google Cloud Functions
  - Azure Functions

Future:
  - Cloud-native processing
  - Serverless MapReduce
  - Event-driven
""")

print("\n=== 3. Unified Processing ===")
print("""
Trend: Batch + Stream unified

Approaches:
  - Apache Beam
  - Spark Structured Streaming
  - Flink

Benefits:
  - One codebase
  - Same logic for batch/stream
  - Easier development

MapReduce role:
  - Batch only
  - Legacy

Future:
  - Unified APIs
  - Lambda architecture simplified
  - Kappa architecture
""")

print("\n=== 4. AI and ML ===")
print("""
Trend: ML on Big Data

MapReduce role:
  - Feature engineering
  - Data preparation
  - Batch inference

Frameworks:
  - Spark MLlib
  - TensorFlow on Spark
  - Horovod

Future:
  - Distributed training
  - Feature stores
  - MLOps
""")

print("\n=== 5. New Architectures ===")
print("""
Trend: Beyond MapReduce

Approaches:
  - Dataflow (Google Cloud Dataflow)
  - Apache Beam
  - Flink (streaming-first)
  - Ray (distributed Python)

Benefits:
  - Better for streaming
  - Better for ML
  - Better for complex pipelines

MapReduce influence:
  - Map and reduce concepts remain
  - Partitioning, shuffle
  - Fault tolerance
""")

print("\n=== 6. Legacy Maintenance ===")
print("""
Current state:
  - Many orgs still run MapReduce
  - Billions of lines of code
  - Critical batch jobs

Challenges:
  - Migration to Spark
  - Skill availability
  - Risk of breaking

Future:
  - Gradual migration
  - Hybrid environments
  - Long tail of legacy
""")

# TIMELINE
print("\n=== Timeline ===")
print("""
Past:
  2004: Google MapReduce paper
  2006: Hadoop (Yahoo)
  2008: Hadoop summit
  2010: Hadoop adoption
  2014: Spark 1.0

Present:
  2020s: Spark dominant
  2025: MapReduce legacy

Future:
  2030: Mostly Spark/Flink?
  2035: New paradigms?
  2040: Unknown
""")

print("\n=== Final Thoughts ===")
print("""
MapReduce is:

  - Foundational for Big Data
  - Simple but powerful
  - Proven at scale
  - Being replaced by Spark

But its concepts live on:
  - Map and reduce
  - Partitioning
  - Shuffle
  - Fault tolerance
  - Data locality

"MapReduce was the first big idea
 in Big Data. It won't be the last."

Keep exploring. Keep building.
""")