MapReduce for Dummies

Say you need to grep for "Hi There" across a petabyte of text spread over thousands of files. A single machine reading that much data sequentially could take hours — I/O alone becomes the bottleneck, before you even count the CPU time to scan every line.

The fix: split the work. Hand different files to different machines (workers) so they scan in parallel, and use a central coordinator to hand out tasks, track progress, and reassign anything that fails.

That's MapReduce in one sentence: split, distribute, collect.

You can find the full implementation of this lab in this commit.

Sample Working example

We use mapReduce to perform grep over 100 files using n=10 workers.

Step 1- Define the coordinator and pass on the tasks

c := Coordinator{
        mu:          sync.Mutex{},
        mapTasks:    make([]Task, len(files)),
        reduceTasks: make([]Task, nReduce),
        nMap:        len(files), //100
        nReduce:     nReduce, //10
    }

for i, file := range files {
     c.mapTasks[i] = Task{
     Id:      i,
     File:    file,}
}

Step 2- Initialize workers and run them until task is complete

// rpc call to co-ordinator
for {
    task, err := fetchTask()
}

func fetchTask() (*Task, error) {
    ok := call("Coordinator.FetchTask", &req, &reply)
    if !ok {
        return &Task{Type: IdleTask}, fmt.Errorf("failed to fetch task")
    }
    return &reply.Task, nil
}

Step 3- Perform Map Task (In this case calling grep)

This step contains the crux of MapReduce -

1. Tasks should be atomic, tasks failed mid-way are discarded
2. Each Map task is divided into N(=10) reduce tasks. So, in total there are map-tasks * NReduce(=10) distinct tasks across nodes.
3. To prevent a stuck/failed worker from hoarding the task, ideally create a deadline which automatically fails the task on crossing it.

content, err := os.ReadFile(task.File)
if err != nil {
    return err
}
if err := c.Err(); err != nil {
    return err
}
// Call map function
kva := grepMyFile(task.File, string(content))
// Write intermediate key-value pairs to files
buckets := make([][]KeyValue, task.NReduce)
for _, kv := range kva {
        // hash ... to be done
    haskKey := ihash(kv.Key) % task.NReduce
    buckets[haskKey] = append(buckets[haskKey], kv)
}

for y, bucket := range buckets {
    // **IMP : create temp file for atomicity
    fileName := fmt.Sprintf("mr-%d-%d", task.Id, y)
    tmpFile, err := os.CreateTemp(".", "mr-tmp-*")

    enc := json.NewEncoder(tmpFile)
    for _, kv := range bucket {
         enc.Encode(&kv)
    }
    tmpFile.Close()
    os.Rename(tmpFile.Name(), fileName)
}

Step 4 - Perform Reduce Task

The reduce Step works in similar way as Map Step. You collect the results produced by various workers to create a unified result.