The Need for MapReduce

Modern applications often require processing massive datasets with low latency. Search engines and recommendation systems must handle extensive user data efficiently. Single machines have prcoessing limitations, leading to the development of distributed computing approaches using interconnected computer clusters. MapReduce builds upon distributed file systems, enabling parallel processing of large-scale computations while handling hardware failures gracefully.

Mapping Phase

The mapping phase transforms input documents into key-value pairs. Consider word frequency counting as an example. For each word occurrence, the mapper emits (word, 1) pairs. This design enables parallel processing by:

• Distributing workload evenly across machines
• Allowing overlapping I/O and computation
• Preventing bottleneck situations

Without mapping, reducers would face unpredictable workloads, causing delays. The mapping phase ensures balanced distribution through hashing techniques.

Reducing Phase

Reducers aggregate mapped results by grouping identical keys. Continuing the word count example, reducers sum values for each word key. Final aggregation combines results from all reducers to produce the complete frequency distribution.

Optimization Techniques

Combiner: Pre-aggregates values locally before shuffling. For repeated keys in a mapper's output (e.g., [(w1,1), (w1,1)]), the combiner emits [(w1,2)] to reduce network traffic.

Shuffling: Distributes mapper outputs evenly across reducers using hash partitioning (hash(key) % reducer_count).

Implementation Example: Wikipedia Word Count

Mapper Implementation

from collections import defaultdict
import os
import re
import pickle
from multiprocessing import Pool

class DocumentProcessor:
    def __init__(self, vocab_path='vocabulary.txt', reducer_count=3):
        with open(vocab_path) as f:
            self.vocab = set(line.strip() for line in f)
        self.reducer_count = reducer_count

    def tokenize(self, filepath):
        with open(filepath, encoding='utf-8') as f:
            text = f.read().lower()
        return re.findall(r'\b\w+\b', text)

    def process_document(self, doc_path):
        tokens = self.tokenize(doc_path)
        counts = defaultdict(int)
        for word in tokens:
            if word in self.vocab:
                doc_id = os.path.basename(doc_path).rsplit('.', 1)[0]
                counts[(doc_id, word)] += 1
        
        partitions = [defaultdict(int) for _ in range(self.reducer_count)]
        for (doc, word), cnt in counts.items():
            partition = hash(word) % self.reducer_count
            partitions[partition][(doc, word)] = cnt
        
        return partitions

    def run_mappers(self, input_dir):
        doc_paths = [os.path.join(root, f) 
                    for root, _, files in os.walk(input_dir) 
                    for f in files]
        
        with Pool() as pool:
            results = pool.map(self.process_document, doc_paths)
        
        for reducer_idx in range(self.reducer_count):
            combined = defaultdict(int)
            for partition in results:
                for k, v in partition[reducer_idx].items():
                    combined[k] += v
            with open(f'map_output_{reducer_idx}.pkl', 'wb') as f:
                pickle.dump(dict(combined), f)

Reducer Implementation

def aggregate_results(reducer_indices):
    for idx in reducer_indices:
        with open(f'map_output_{idx}.pkl', 'rb') as f:
            mapped_data = pickle.load(f)
        
        word_totals = defaultdict(int)
        doc_index = defaultdict(set)
        
        for (doc_id, word), count in mapped_data.items():
            word_totals[word] += count
            doc_index[doc_id].add(word)
        
        sorted_words = sorted(word_totals.items(), 
                             key=lambda x: x[1], 
                             reverse=True)
        
        with open(f'reduced_{idx}.pkl', 'wb') as f:
            pickle.dump({
                'word_counts': dict(sorted_words),
                'document_index': dict(doc_index)
            }, f)