Yingjie's Blog

Implementing MapReduce in Java

Introduction

MapReduce is a distributed computing model designed for processing large datasets. It divides data processing tasks into Map and Reduce phases, achieving computational efficiency through parallel processing. This article will demonstrate how to implement a simple MapReduce framework in Java, similar to the functionality described in Google's MapReduce paper.

graph LR
A[Input Data] --> B[Split]
B --> C1[Map 1]
B --> C2[Map 2]
B --> C3[Map 3]
C1 --> D[Shuffle]
C2 --> D
C3 --> D
D --> E1[Reduce 1]
D --> E2[Reduce 2]
E1 --> F[Output Result]
E2 --> F

System Architecture

Our Java implementation uses a Master-Worker architecture:

graph TD
A[Master] --> B1[Worker 1]
A --> B2[Worker 2]
A --> B3[Worker 3]
B1 --> C1[Map Task]
B1 --> C2[Reduce Task]
B2 --> C3[Map Task]
B2 --> C4[Reduce Task]
B3 --> C5[Map Task]
B3 --> C6[Reduce Task]

The main components include:

  1. Master: Responsible for task assignment and scheduling
  2. Worker: Executes specific Map/Reduce tasks
  3. Task: Represents information about a Map or Reduce task

Core Implementation

KeyValue Class

The KeyValue class represents key-value pairs:

public class KeyValue implements Comparable<KeyValue> {
    public String key;
    public String value;
    
    public KeyValue(String key, String value) {
        this.key = key;
        this.value = value;
    }
    
    @Override
    public int compareTo(KeyValue o) {
        return key.compareTo(o.key);
    }
}

Map Process

Map phase workflow:

sequenceDiagram
    participant Worker
    participant MapF
    participant File System
    
    Worker->>MapF: doMap(task)
    MapF->>File System: Read input file
    MapF->>MapF: Process data
    MapF->>File System: Write intermediate results
    MapF->>Worker: Complete Map task

Map function implementation:

public static List<KeyValue> sequentialMap(String filename, String content) {
    Pattern pattern = Pattern.compile("[a-zA-Z]+");
    Matcher matcher = pattern.matcher(content);
    List<String> words = new ArrayList<>();
    while (matcher.find()) {
        words.add(matcher.group());
    }
    List<KeyValue> result = new ArrayList<>();
    for (String word : words) {
        KeyValue kv = new KeyValue(word, "1");
        result.add(kv);
    }
    return result;
}

Reduce Process

Reduce phase workflow:

sequenceDiagram
    participant Worker
    participant ReduceF
    participant File System
    
    Worker->>ReduceF: doReduce(task)
    ReduceF->>File System: Read Map output
    ReduceF->>ReduceF: Merge processing
    ReduceF->>File System: Write final results
    ReduceF->>Worker: Complete Reduce task

Reduce function implementation:

public static String sequentialReduce(String key, List<String> values) {
    return String.valueOf(values.size());
}

Task Scheduling

The Master is responsible for task assignment and scheduling, with the following core workflow:

stateDiagram-v2
    [*] --> Idle
    Idle --> MapTask: Has Map tasks
    MapTask --> MapTask: Task in progress
    MapTask --> ReduceTask: Map complete
    ReduceTask --> ReduceTask: Task in progress
    ReduceTask --> Complete: All tasks complete
    Complete --> [*]

Data Flow Between Map and Reduce

The data flow between Map task outputs and Reduce task inputs is a crucial part of the MapReduce framework. In our implementation, this process includes several important steps:

  1. Partition Writing

After processing data, Map tasks write results to temporary files on local disk. To ensure data is correctly distributed to different Reduce tasks, we use a partitioning strategy:

// Partition writing logic in MapF.java
for (KeyValue kv : kvl) {
    // Use key's hash value modulo number of Reduce tasks to determine partition
    intermediate.get(hash(kv.key) % task.getnReduce()).add(kv);
}

// Write each partition's data to corresponding temporary file
for (int i = 0; i < task.getnReduce(); i++) {
    String oName = "mr-%d-%d".formatted(task.getTaskID(), i);
    File oFile = new File(oName);
    FileWriter fw = new FileWriter(oFile, true);
    for (KeyValue kv : intermediate.get(i)) {
        fw.append(template.formatted(kv.key, kv.value));
    }
    fw.close();
}

The key points about temporary file naming:

  • mr-X-Y: X is the Map task ID, Y is the target Reduce task ID
  • This naming convention ensures each Reduce task can accurately find its data
  1. File Organization
graph TD
    M1[Map Task 1] --> F1[mr-1-0]
    M1 --> F2[mr-1-1]
    M2[Map Task 2] --> F3[mr-2-0]
    M2 --> F4[mr-2-1]
    F1 --> R1[Reduce Task 0]
    F3 --> R1
    F2 --> R2[Reduce Task 1]
    F4 --> R2
  1. Reduce Task Data Collection

When a Reduce task starts, it collects all temporary files generated for it by Map tasks:

// Data collection logic in ReduceF.java
List<KeyValue> intermediate = new ArrayList<>();
for (int i = 0; i < task.getnMap(); i++) {
    // Construct corresponding temporary filename
    String iName = "mr-%d-%d".formatted(i, task.getTaskID());
    File iFile = new File(iName);
    
    // Read file contents
    try (FileInputStream fs = new FileInputStream(iFile)) {
        byte[] data = new byte[(int) iFile.length()];
        fs.read(data);
        String[] lines = new String(data, StandardCharsets.UTF_8).split("\n");
        for (String line : lines) {
            String[] content = line.split(" ");
            intermediate.add(new KeyValue(content[0], content[1]));
        }
    }
}

// Sort collected data
Collections.sort(intermediate);
  1. Data Merge Processing

After collecting data, the Reduce task merges data with the same key:

int i = 0;
while (i < intermediate.size()) {
    int j = i + 1;
    // Find range of data with same key
    while (j < intermediate.size() && 
           intermediate.get(j).key.equals(intermediate.get(i).key)) {
        j++;
    }
    
    // Collect all values for the same key
    List<String> values = new LinkedList<>();
    for (int k = i; k < j; k++) {
        values.add(intermediate.get(k).value);
    }
    
    // Call reduce function for processing
    String output = reduceF.apply(intermediate.get(i).key, values);
    
    // Write final results
    try (FileWriter fw = new FileWriter(oFile, true)) {
        String content = "%s %s\n".formatted(intermediate.get(i).key, output);
        fw.append(content);
    }
    
    i = j;
}

The key points of this process are:

  1. Using hash partitioning to ensure data with the same key goes to the same Reduce task
  2. Using standardized file naming to ensure correct data flow
  3. Sorting data in the Reduce phase for easier grouping
  4. Automatic management of temporary files (creation and cleanup) by the framework

This design ensures data flows correctly from Map tasks to Reduce tasks while supporting fault tolerance: if a task fails, only the corresponding Map or Reduce task needs to be re-executed.

Usage Example

  1. Start the Master server:
public static void main(String[] args) {
    Master master = new MasterImpl(readFiles(), 10);
    Naming.bind("rmi://localhost:1099/master", master);
    System.out.println("RMI server started.");
}
  1. Start a Worker:
public static void main(String[] args) {
    Worker worker = new Worker();
    worker.work(MapF::sequentialMap, ReduceF::sequentialReduce);
}

Performance Optimization

  1. Task Granularity
  • Set appropriate number of Map and Reduce tasks
  • Avoid too many tasks causing scheduling overhead
  • Avoid too few tasks preventing full parallelism utilization
  1. Data Locality
  • Try to have Workers process local data
  • Reduce network transfer overhead
  1. Fault Tolerance
  • Timeout retry mechanism
  • Failed task reassignment

Summary

This article demonstrated how to implement a simple MapReduce framework in Java. Using a Master-Worker architecture, we implemented distributed computing supporting parallel processing of large datasets. Key points include:

  1. Two-phase Map/Reduce processing
  2. Distributed task scheduling
  3. Fault tolerance mechanism
  4. Parallel computation

While this implementation is simple, it contains the core concepts and basic workflow of MapReduce, helping understand MapReduce principles.

References

  1. Dean, Jeffrey, and Sanjay Ghemawat. "MapReduce: simplified data processing on large clusters." Communications of the ACM 51.1 (2008): 107-113.
  2. Kleppmann, Martin. "Designing Data-Intensive Applications: The Big Ideas Behind Reliable, Scalable, and Maintainable Systems." O'Reilly Media, 2017.

Licensed under CC BY-NC-SA 4.0