REDUCE FUNCTION SIGNATURE
────────────────────────

reduce(K2 key, Iterator<V2> values) → list<K3, V3>

Input:
• K2: Intermediate key (from map)
• values: ALL values for this key (grouped by framework)

Output:
• Emit zero or more (K3, V3) pairs
• Typically K3 = K2, V3 is aggregated result


EXAMPLE: Word Count Reduce
──────────────────────────

Input: ("hello", [1, 1, 1, 1, 1])

Process:
1. Sum all values: 1 + 1 + 1 + 1 + 1 = 5
2. Emit result: emit("hello", 5)

Output: ("hello", 5)


Another Input: ("world", [1, 1])

Process:
1. Sum: 1 + 1 = 2
2. Emit: emit("world", 2)

Output: ("world", 2)


KEY INSIGHTS:
────────────
• Reducer receives ALL values for a key
• Keys are sorted before reduce
• Reducers run in parallel (different keys)
• Each reducer processes multiple keys
• Values arrive as an iterator (may not fit in memory)

public class WordCountReducer
    extends Reducer<Text, IntWritable, Text, IntWritable> {

  @Override
  public void reduce(Text key, Iterable<IntWritable> values,
                    Context context)
      throws IOException, InterruptedException {

    int sum = 0;
    for (IntWritable val : values) {
      sum += val.get();
    }

    context.write(key, new IntWritable(sum));
  }
}

public class WordCount {

  // Map Phase
  public static class TokenizerMapper
      extends Mapper<LongWritable, Text, Text, IntWritable> {

    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 {
      StringTokenizer itr = new StringTokenizer(value.toString());
      while (itr.hasMoreTokens()) {
        word.set(itr.nextToken());
        context.write(word, one);
      }
    }
  }

  // Reduce Phase
  public static class IntSumReducer
      extends Reducer<Text, IntWritable, Text, IntWritable> {

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

  // Main: Configure and submit job
  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(TokenizerMapper.class);
    job.setReducerClass(IntSumReducer.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);
  }
}

Input File (input.txt):
───────────────────────
hello world
hello hadoop
world mapreduce


Map Phase Output (intermediate):
─────────────────────────────────
("hello", 1)
("world", 1)
("hello", 1)
("hadoop", 1)
("world", 1)
("mapreduce", 1)


After Shuffle & Sort (grouped by key):
──────────────────────────────────────
("hadoop", [1])
("hello", [1, 1])
("mapreduce", [1])
("world", [1, 1])


Reduce Phase Output (final):
────────────────────────────
hadoop      1
hello       2
mapreduce   1
world       2

MAP OUTPUT BUFFER
────────────────

Configuration: io.sort.mb = 100 (default 100MB)

┌─────────────────────────────────────┐
│      Circular Memory Buffer         │
│           (100MB)                   │
├─────────────────────────────────────┤
│ (key1, val1, partition1, metadata)  │
│ (key2, val2, partition2, metadata)  │
│ (key3, val3, partition3, metadata)  │
│ ...                                 │
│ ...                                 │
└─────────────────────────────────────┘
     │
     │ When 80% full (configurable)
     ↓
BACKGROUND SPILL TO DISK
───────────────────────

1. Sort by (partition, key):
   - Primary sort: partition ID
   - Secondary sort: key

2. Optionally run Combiner (mini-reduce):
   - Local aggregation to reduce data
   - E.g., for word count: ("hello",[1,1,1]) → ("hello",3)

3. Write sorted partitions to disk:

   spill_0.out:
   ┌──────────────────────────────┐
   │ Partition 0 (sorted by key)  │
   │   ("apple", 2)               │
   │   ("banana", 1)              │
   ├──────────────────────────────┤
   │ Partition 1 (sorted by key)  │
   │   ("cherry", 3)              │
   │   ("date", 1)                │
   ├──────────────────────────────┤
   │ Partition 2 (sorted by key)  │
   │   ("elderberry", 1)          │
   │   ("fig", 2)                 │
   └──────────────────────────────┘


MULTIPLE SPILLS:
───────────────

If map produces lots of output:
- spill_0.out
- spill_1.out
- spill_2.out
- ...

These must be merged before shuffle.


FINAL MERGE:
───────────

Merge all spill files into single output file:
- Multi-way merge (up to io.sort.factor files at once)
- Maintains sort order
- Run combiner again if configured
- Result: Single sorted file per map task

COMBINER OPTIMIZATION
────────────────────

Problem:
• Map emits many duplicate keys
• Lots of network transfer during shuffle
• Reducers receive massive input

Example (Word Count without Combiner):
─────────────────────────────────────

Mapper 1 output:
("hello", 1), ("hello", 1), ("hello", 1), ("world", 1)

Mapper 2 output:
("hello", 1), ("hello", 1), ("world", 1), ("world", 1)

Shuffle transfers: 8 (key, value) pairs


Solution: COMBINER (local aggregation)
──────────────────────────────────────

Combiner runs on map node BEFORE shuffle:

Mapper 1 output after combiner:
("hello", 3), ("world", 1)

Mapper 2 output after combiner:
("hello", 2), ("world", 2)

Shuffle transfers: 4 (key, value) pairs
→ 50% reduction!


COMBINER FUNCTION:
─────────────────

Often SAME as reducer function:

public class WordCountCombiner
    extends Reducer<Text, IntWritable, Text, IntWritable> {

  @Override
  public void reduce(Text key, Iterable<IntWritable> values,
                    Context context)
      throws IOException, InterruptedException {
    int sum = 0;
    for (IntWritable val : values) {
      sum += val.get();
    }
    context.write(key, new IntWritable(sum));
  }
}

// Configure job
job.setCombinerClass(WordCountCombiner.class);


WHEN TO USE COMBINER:
────────────────────

✓ Associative and commutative operations:
  - Sum, count, max, min
  - reduce(reduce(A, B), C) = reduce(A, reduce(B, C))

✗ NOT suitable for:
  - Average (need count and sum separately)
  - Median, percentiles
  - Operations where order matters


COMBINER GUARANTEES:
───────────────────

• May run 0, 1, or multiple times
• Framework decides when to run
• Your code must work with or without combiner
• Must produce same result regardless

SHUFFLE INTERNALS: MAP-SIDE
────────────────────────────

1. Memory Buffer (io.sort.mb, default 100MB):
- Circular buffer where (key,value) pairs are initially written
- When buffer reaches `io.sort.spill.percent` (default 80%), a "spill thread" kicks in
- Spill sorts by (partition, key) using quicksort
- If combiner is configured, it runs here to reduce output

2. Spill to Disk:
- Sorted data is written to a spill file: `map-00001.out`
- Each spill file contains data for all partitions, but sorted within each partition
- Multiple spills may occur if the map task processes large amounts of data

3. Final Merge:
- Before the map task finishes, all spill files are merged into a single output file
- This is a multi-way merge, respecting the `io.sort.factor` (default 10) limit on simultaneous merges
- The final file is named `map-00001.out` and stored locally on the DataNode

4. Index File:
- Alongside the output file, an index file `map-00001.out.index` is created
- This index maps each partition ID to its location and length in the output file
- Crucial for the reduce-side fetch process

SHUFFLE INTERNALS: REDUCE-SIDE
──────────────────────────────

1. Fetch Initiation:
- The reduce task starts fetching map outputs as soon as any map task completes
- It connects to the HTTP server running on each TaskTracker that ran a map task
- Fetches are performed in parallel, up to `mapreduce.reduce.shuffle.parallelcopies` (default 5)

2. In-Memory Buffering:
- Fetched data is kept in a memory buffer (size controlled by `mapreduce.reduce.shuffle.input.buffer.percent`)
- If data arrives faster than it can be processed, it spills to disk

3. On-Disk Merge:
- Multiple fetched files are merged on the reducer's local disk
- This uses the same multi-way merge algorithm as the map-side final merge
- Result is a single, sorted file per reduce task

4. Final Sort and Group:
- All data is sorted by key (secondary sort during merge ensures this)
- Values for each key are grouped together
- The user's reduce function is called for each key-group

PERFORMANCE BOTTLENECKS:
──────────────────────
- Network Bandwidth: Shuffle is often network-bound
- Disk I/O: Writing/reading spill files can be slow
- Memory Pressure: Both map and reduce sides need adequate heap
- Skewed Keys: A few keys with disproportionately large value lists can slow the entire reduce task

MERGE PHASE
──────────

Reducer has fetched outputs from ALL mappers:

In Memory:
• 20 small files (each sorted)

On Disk:
• 50 larger files (each sorted)

Goal: Merge into single sorted stream


MULTI-LEVEL MERGE:
─────────────────

Configuration: io.sort.factor = 10
(merge up to 10 files at once)

Round 1: Merge on-disk files
┌─────────────────────────────┐
│ Merge files 1-10  → merged1 │
│ Merge files 11-20 → merged2 │
│ Merge files 21-30 → merged3 │
│ Merge files 31-40 → merged4 │
│ Merge files 41-50 → merged5 │
└─────────────────────────────┘

Round 2:
┌─────────────────────────────┐
│ Merge merged1-merged5       │
│   + in-memory files         │
│   → Final sorted stream     │
└─────────────────────────────┘


MERGE ALGORITHM (K-way merge):
──────────────────────────────

Use min-heap to efficiently merge K sorted files:

Input Files:
File 1: [1, 5, 9, 13]
File 2: [2, 6, 10, 14]
File 3: [3, 7, 11, 15]
File 4: [4, 8, 12, 16]

Min-Heap (track next element from each file):
┌───────────────────────────┐
│ (1, file1) ← root (min)   │
│ (2, file2)                │
│ (3, file3)                │
│ (4, file4)                │
└───────────────────────────┘

Algorithm:
1. Pop min from heap (1, file1)
2. Output: 1
3. Read next from file1 (5), insert to heap
4. Repeat...

Result: [1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16]

Time Complexity: O(N log K)
where N = total elements, K = number of files


FINAL SORTED OUTPUT:
───────────────────

(key, [values]) grouped by key:

("apple", [1, 2, 1, 3, 1])
("banana", [1, 1, 4, 1])
("cherry", [2, 1, 1])
...

Ready for reduce() function!

REDUCE EXECUTION
───────────────

Input: Sorted, grouped (key, values) pairs

For each unique key:
  Call user's reduce(key, values) function


Example (Word Count):
────────────────────

Input stream:
("apple", [1, 2, 1, 3, 1])
("banana", [1, 1, 4, 1])
("cherry", [2, 1, 1])

Execution:
─────────
reduce("apple", [1,2,1,3,1])
  → sum = 8
  → emit("apple", 8)

reduce("banana", [1,1,4,1])
  → sum = 7
  → emit("banana", 7)

reduce("cherry", [2,1,1])
  → sum = 4
  → emit("cherry", 4)


OUTPUT WRITING:
──────────────

Reduce outputs written to HDFS:

/user/hadoop/output/
  ├── part-r-00000  (from Reducer 0)
  ├── part-r-00001  (from Reducer 1)
  ├── part-r-00002  (from Reducer 2)
  └── _SUCCESS      (marker file)

Each reducer writes ONE output file.


OUTPUT FORMAT:
─────────────

Default: TextOutputFormat
Format: key\tvalue\n

part-r-00000:
──────────────
apple	8
banana	7
cherry	4
...


CUSTOM OUTPUT FORMAT:
────────────────────

SequenceFile (binary, efficient):
job.setOutputFormatClass(SequenceFileOutputFormat.class);

Multiple Outputs (different files/formats):
MultipleOutputs.addNamedOutput(job, "text",
                              TextOutputFormat.class,
                              Text.class, IntWritable.class);

No Reducer (map-only job):
job.setNumReduceTasks(0);
// Maps write directly to HDFS

WHY COMPRESS:
────────────

✓ Less disk I/O (spills, shuffle)
✓ Less network transfer
✓ Faster job completion
✗ CPU overhead (compression/decompression)


COMPRESSION CODECS:
──────────────────

┌──────────────┬──────────┬──────────┬───────────┐
│ Codec        │ Ratio    │ Speed    │ Splittable│
├──────────────┼──────────┼──────────┼───────────┤
│ Gzip         │ High     │ Slow     │ No        │
│ Bzip2        │ Highest  │ Slowest  │ Yes       │
│ LZO          │ Medium   │ Fast     │ Yes*      │
│ Snappy       │ Medium   │ Fastest  │ No        │
└──────────────┴──────────┴──────────┴───────────┘

* LZO requires indexing for splittability


WHERE TO COMPRESS:
─────────────────

1. Input Files:
   - Reduces map input I/O
   - Must be splittable or small files
   - Configure: InputFormat handles automatically

2. Map Output (Intermediate):
   - Reduces shuffle traffic (BIG win!)
   - CPU overhead acceptable

   Configuration:
   mapreduce.map.output.compress=true
   mapreduce.map.output.compress.codec=
     org.apache.hadoop.io.compress.SnappyCodec

3. Final Output:
   - Saves HDFS space
   - Slows down if consumed by non-Hadoop tools

   Configuration:
   mapreduce.output.fileoutputformat.compress=true
   mapreduce.output.fileoutputformat.compress.codec=
     org.apache.hadoop.io.compress.GzipCodec


RECOMMENDATION:
──────────────

ALWAYS compress intermediate data (map output):
→ 30-50% reduction in shuffle time
→ Minimal CPU overhead with Snappy/LZO

Example:
────────
Configuration conf = new Configuration();
conf.setBoolean("mapreduce.map.output.compress", true);
conf.setClass("mapreduce.map.output.compress.codec",
              SnappyCodec.class, CompressionCodec.class);


COMPRESSION IMPACT:
──────────────────

Uncompressed shuffle: 100 GB
Snappy compressed: 40 GB
Network time (1 Gbps): 100s → 40s (60% faster!)

COMBINER EFFECTIVENESS:
──────────────────────

Word Count Example:
──────────────────

WITHOUT Combiner:
Map output: 10 million (word, 1) pairs
Shuffle: 10 million pairs transferred
Reduce input: 10 million pairs

WITH Combiner:
Map output: 10 million (word, 1) pairs
After combiner: 100,000 (word, count) pairs
Shuffle: 100,000 pairs transferred (99% reduction!)
Reduce input: 100,000 pairs


WHEN COMBINER HELPS:
───────────────────

High Reduction Factor:
• Many duplicate keys
• Simple aggregation (sum, count, max)
• Word count, click counting, log aggregation

Low Reduction Factor:
• Few duplicate keys
• Unique IDs, timestamps
• Combiner overhead not worth it


COMBINER USAGE PATTERNS:
───────────────────────

1. Same as Reducer (common):
   job.setCombinerClass(MyReducer.class);

2. Different from Reducer:
   job.setCombinerClass(MyCombiner.class);
   job.setReducerClass(MyReducer.class);

3. In-Mapper Combining (advanced):
   // Maintain HashMap in mapper
   // Emit aggregated results in cleanup()


CONFIGURATION:
─────────────

mapreduce.job.combine.class = MyReducer


COMBINER FREQUENCY:
──────────────────

Combiner runs:
• During spills (map side)
• During merge (map side)
• Possibly during shuffle merge (reduce side)

→ May run 0, 1, or multiple times!
→ Must be idempotent and associative

PARTITIONING STRATEGIES:
───────────────────────

1. Hash Partitioning (Default):
   partition = hash(key) % numReducers

   Pros: Simple, uniform for random keys
   Cons: Poor for skewed keys


2. Range Partitioning:
   partition based on key range

   Example:
   • Reducer 0: A-F
   • Reducer 1: G-M
   • Reducer 2: N-S
   • Reducer 3: T-Z

   Pros: Output is globally sorted
   Cons: Requires sampling, skew issues


3. Custom Partitioning:

   public class FirstLetterPartitioner
       extends Partitioner<Text, IntWritable> {

     @Override
     public int getPartition(Text key, IntWritable value,
                            int numPartitions) {
       // Partition by first letter
       char first = key.toString().toLowerCase().charAt(0);
       if (first >= 'a' && first <= 'm') {
         return 0 % numPartitions;
       } else {
         return 1 % numPartitions;
       }
     }
   }


SKEW PROBLEM:
────────────

Unbalanced partitions:

Reducer 0: 1 GB  (finishes fast)
Reducer 1: 10 GB (straggles)
Reducer 2: 2 GB
Reducer 3: 1 GB

Job completion time = slowest reducer


SKEW SOLUTIONS:
──────────────

1. Better Partitioner:
   • Sample data first
   • Learn key distribution
   • Assign ranges to balance load

2. Increase Number of Reducers:
   • More parallelism
   • Smaller data per reducer
   • job.setNumReduceTasks(20);

3. Salting:
   • Add random suffix to skewed keys
   • key → key_0, key_1, key_2, ...
   • Process in two MapReduce rounds

4. Combiners:
   • Reduce data before shuffle
   • Less data to partition

Google's Approach:
─────────────────
• Publish papers, keep code internal
• Full control over evolution
• Optimize for Google's workloads
• No backward compatibility burden

Hadoop's Approach:
─────────────────
• Open source from day one
• Community-driven development
• Serve diverse use cases
• Maintain backward compatibility

Impact:
──────
• Hadoop became industry standard
• Google's ideas reached millions
• Ecosystem innovations (Spark, Hive)
• Multiple commercial distributions

Google's Evolution (Opaque):
───────────────────────────
MapReduce (2004)
  ↓
Flume (streaming, internal)
  ↓
MillWheel (streaming)
  ↓
Cloud Dataflow (public product)

Likely deprecated internally, replaced by newer systems.


Hadoop's Evolution (Transparent):
─────────────────────────────────
MapReduce (2006)
  ↓
YARN (2012) - Generic resource management
  ↓
Multiple frameworks: Spark, Flink, Tez
  ↓
Modern: Spark dominates, MapReduce legacy

Every step documented and community-driven.


Lessons:
───────
• Open evolution allows external innovation
• Community found MapReduce limitations
• Led to better alternatives (Spark)
• Google's internal evolution hidden