writing

MapReduce From Scratch in Go

15 min read

MapReduce is one of those ideas that sounds simple in a sentence and gets genuinely interesting the moment you try to make it survive real failures. I built a working implementation — coordinator, workers, RPC, crash recovery, the whole thing — against the MIT 6.5840 distributed systems lab interface, and this is a walk through how it's put together, why the pieces are shaped the way they are, and where I actually broke it on purpose to check the design held up.

Repo: mapreduce-go (Go 1.22, no external dependencies beyond the standard library).

Before writing any code, I spent time just reasoning through the problem without touching a keyboard — working out why a worker can't just Map-then-Reduce its own file in isolation, why the intermediate file count is M × nReduce, what breaks if two copies of the same task both finish. That ordering matters. Distributed systems bugs are almost never syntax errors; they're the consequence of a design decision you didn't think hard enough about before writing it down. This post follows roughly the same order I actually worked through it.

What MapReduce actually is

Strip away the "big data" framing and MapReduce is a pattern for one specific problem: you have a large dataset spread across many files, and you want to compute something over it — word counts, an inverted index, log aggregation, whatever — faster than one process reading everything sequentially could manage. Google's original 2004 paper framed it as a way to let engineers write simple sequential code and get parallelism and fault tolerance "for free," without having to think about network programming or crash recovery themselves. That "for free" part is a promise made to the caller of MapReduce — someone underneath still has to build the coordinator and worker machinery that makes it true, and that's the part I actually built.

The trick is splitting the computation into two functions with very different shapes:

  • Map(filename, contents) → [(key, value), ...] runs independently per input file. No shared state, no coordination needed between mappers. Perfectly parallel.
  • Reduce(key, [values]) → output runs once per distinct key, over every value that key ever produced across all map calls.

The interesting part isn't either function — it's the step in between, usually called shuffle: getting every value for a given key routed to the same reducer, when map tasks ran independently and don't know about each other. That's the part that turns "two simple functions" into a distributed systems problem, and it's where almost all of the actual engineering in this project lives.

The shape of the system

It helps to first look at the non-distributed version, because it's the spec you're trying to match. The lab ships a mrsequential.go reference: read every input file, call Map on each, sort and group the combined output by key, call Reduce on each group, write results. One process, one loop, correct by construction. The entire job of the distributed version is to produce byte-identical output to that, while splitting the work across many processes that can crash independently.

Two kinds of processes make that happen:

  • One coordinator — hands out work, tracks progress, decides when the job is done.
  • Many workers — ask the coordinator for a task, execute it, report back, ask for the next one.

Workers never talk to each other. All coordination flows through the coordinator over RPC. This is deliberate: it keeps the failure model simple. A worker can die at any point and the only thing that needs to notice is the coordinator — not some peer worker that also has to detect the failure and agree with everyone else about what happened.

The transport is a Unix domain socket rather than TCP, since every process runs on the same machine:

func coordinatorSock() string {
	s := "/var/tmp/5840-mr-"
	s += strconv.Itoa(os.Getuid())
	return s
}

The coordinator registers itself and listens on that socket:

func (c *Coordinator) server() {
	rpc.Register(c)
	rpc.HandleHTTP()
	sockname := coordinatorSock()
	os.Remove(sockname)
	l, e := net.Listen("unix", sockname)
	if e != nil {
		log.Fatal("listen error:", e)
	}
	go http.Serve(l, nil)
}

Go's net/rpc package handles the serialization and dispatch — any exported method on Coordinator with the right signature (func(args *T, reply *U) error) becomes callable by name over the socket. That's the entire RPC layer; no protobuf, no custom wire format, just Go's standard library doing what it's built for.

Two RPCs cover the whole protocol:

type GetTaskArgs struct{}

type GetTaskReply struct {
	TaskType  TaskType
	TaskId    int
	InputFile string // only meaningful for MapTask
	NReduce   int
	NMap      int
}

type ReportTaskArgs struct {
	TaskType TaskType
	TaskId   int
}

type ReportTaskReply struct{}

GetTaskArgs is empty — a worker isn't telling the coordinator anything when it asks for work, it's purely requesting. ReportTaskArgs carries just enough for the coordinator to know which task to mark done.

Four possible values for TaskType: MapTask, ReduceTask, WaitTask (nothing assignable right now, but the job isn't finished — sleep briefly and ask again), and ExitTask (job's done, shut down). The worker's entire control loop is a switch on this one field:

for {
	reply := GetTaskReply{}
	ok := call("Coordinator.GetTask", &GetTaskArgs{}, &reply)
	if !ok {
		return // coordinator is gone — job's done, nothing left to do
	}

	switch reply.TaskType {
	case MapTask:
		doMap(mapf, reply.TaskId, reply.InputFile, reply.NReduce)
		call("Coordinator.ReportTask", &ReportTaskArgs{TaskType: MapTask, TaskId: reply.TaskId}, &ReportTaskReply{})
	case ReduceTask:
		doReduce(reducef, reply.TaskId, reply.NMap)
		call("Coordinator.ReportTask", &ReportTaskArgs{TaskType: ReduceTask, TaskId: reply.TaskId}, &ReportTaskReply{})
	case WaitTask:
		time.Sleep(200 * time.Millisecond)
	case ExitTask:
		return
	}
}

Notice the worker doesn't distinguish "coordinator unreachable" from "job finished, coordinator exited" — both just mean stop. That's a deliberate simplification: in this system, the coordinator's process lifetime is the job's lifetime, so a dead coordinator and a finished job look identical from a worker's point of view.

Partitioning: how values find their way to the right reducer

Say you have M map tasks and want nReduce reducers. Each mapper doesn't write one output file — it writes nReduce of them, bucketing every key/value pair it produces by hash(key) % nReduce:

func ihash(key string) int {
	h := fnv.New32a()
	h.Write([]byte(key))
	return int(h.Sum32() & 0x7fffffff)
}

buckets := make([][]KeyValue, nReduce)
for _, kv := range kva {
	b := ihash(kv.Key) % nReduce
	buckets[b] = append(buckets[b], kv)
}

Each bucket gets written to a file named mr-<mapTaskId>-<bucket>. Run that across M map tasks and you end up with M × nReduce intermediate files total — a small grid where rows are map tasks and columns are reduce buckets.

Here's why the naming matters: reduce task Y doesn't need to know anything about which map tasks succeeded, retried, or in what order — it just reads every file matching mr-*-Y, one contribution from each mapper, and it's guaranteed to have every value for every key that hashed into bucket Y. The filename is the routing table.

func doReduce(reducef func(string, []string) string, taskId int, nMap int) {
	var kva []KeyValue
	for m := 0; m < nMap; m++ {
		inname := fmt.Sprintf("mr-%d-%d", m, taskId)
		file, err := os.Open(inname)
		if err != nil {
			if os.IsNotExist(err) {
				continue // this mapper just produced nothing for this bucket
			}
			log.Fatalf("cannot open %v: %v", inname, err)
		}
		dec := json.NewDecoder(file)
		for {
			var kv KeyValue
			if err := dec.Decode(&kv); err != nil {
				break
			}
			kva = append(kva, kv)
		}
		file.Close()
	}
	sort.Sort(byKey(kva))
	// group adjacent equal keys, call reducef once per group...
}

Sorting first means every occurrence of a key ends up adjacent, so grouping is a single linear pass — bump a pointer forward while keys match, hand the whole slice of values to reducef, repeat.

Swapping applications without recompiling the worker

One detail that trips people up the first time: mrworker doesn't have Map and Reduce compiled into it. It loads them at runtime from a .so file using Go's plugin package:

func loadPlugin(filename string) (func(string, string) []mr.KeyValue, func(string, []string) string) {
	p, err := plugin.Open(filename)
	if err != nil {
		log_fatalf("cannot load plugin %v: %v", filename, err)
	}
	xmapf, err := p.Lookup("Map")
	if err != nil {
		log_fatalf("cannot find Map in %v: %v", filename, err)
	}
	mapf := xmapf.(func(string, string) []mr.KeyValue)

	xreducef, err := p.Lookup("Reduce")
	if err != nil {
		log_fatalf("cannot find Reduce in %v: %v", filename, err)
	}
	reducef := xreducef.(func(string, []string) string)

	return mapf, reducef
}

The application — word count, an inverted index builder, whatever — is a separate small Go file exporting Map and Reduce, compiled with go build -buildmode=plugin. Swap the .so you pass to mrworker and you've got a completely different job, with zero changes to the coordinator or the worker's scheduling logic. It's a clean separation: the worker owns how work gets distributed and executed safely; the plugin owns what the computation actually is. Worth knowing if you're on Windows, though — plugin only supports Linux and macOS, so this whole approach needs WSL there.

The coordinator: task states, phases, and the timeout that makes this actually distributed

The naive version of a coordinator is a queue: hand out tasks, mark them done when reported. That works until a worker dies mid-task and never reports back — now that task is stuck forever and the job never finishes.

The coordinator tracks each task with a small state machine — idle, inProgress, or completed — plus, crucially, a timestamp of when it was last handed out:

type taskState int

const (
	idle taskState = iota
	inProgress
	completed
)

type taskBookkeeping struct {
	state     taskState
	startTime time.Time
}

The whole job also moves through three phases — mapPhase, reducePhase, donePhase — and the coordinator refuses to hand out reduce tasks until every map task is completed. This isn't arbitrary: reduce tasks read mr-X-Y files that map tasks produce, so starting reduce early would mean reading intermediate files that don't fully exist yet.

GetTask is really just "look at the current phase, try to find something assignable in that phase's task list":

func (c *Coordinator) GetTask(args *GetTaskArgs, reply *GetTaskReply) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	switch c.phase {
	case mapPhase:
		if id, ok := c.findAssignableTask(c.mapTasks); ok {
			c.mapTasks[id].state = inProgress
			c.mapTasks[id].startTime = time.Now()
			reply.TaskType = MapTask
			reply.TaskId = id
			reply.InputFile = c.files[id]
			reply.NReduce = c.nReduce
			reply.NMap = c.nMap
			return nil
		}
		reply.TaskType = WaitTask
		return nil

	case reducePhase:
		if id, ok := c.findAssignableTask(c.reduceTasks); ok {
			c.reduceTasks[id].state = inProgress
			c.reduceTasks[id].startTime = time.Now()
			reply.TaskType = ReduceTask
			reply.TaskId = id
			reply.NReduce = c.nReduce
			reply.NMap = c.nMap
			return nil
		}
		reply.TaskType = WaitTask
		return nil

	default: // donePhase
		reply.TaskType = ExitTask
		return nil
	}
}

Everything is behind a single mutex. RPC handlers in Go's net/rpc run concurrently — multiple workers can call GetTask at the same instant — so without the lock, two workers could both see the same task as idle and both get assigned it, which defeats the entire point of task tracking. One mutex over a handful of slices is enough here; the state being protected is small and the critical sections are short, so there's no real contention to worry about.

findAssignableTask is where the timeout logic actually lives — and it's doing double duty, looking for genuinely untouched work and work that's been abandoned:

func (c *Coordinator) findAssignableTask(tasks []taskBookkeeping) (int, bool) {
	for i := range tasks {
		if tasks[i].state == idle {
			return i, true
		}
		if tasks[i].state == inProgress && time.Since(tasks[i].startTime) > taskTimeout {
			return i, true
		}
	}
	return 0, false
}

Ten seconds, in this implementation. No heartbeats, no "I'm still alive" messages from workers — just: if you took too long, we assume you're gone and give the work to someone else. This is the single idea that turns a task scheduler into something that survives real crashes, and it's a much simpler mechanism than it sounds — there's no failure detector process, no gossip, no consensus. The coordinator just checks a clock every time someone asks for work.

The consequence: two workers can end up completing the same task — the original one, slow but not actually dead, and the one that got the reassignment. ReportTask has to be idempotent about this:

func (c *Coordinator) ReportTask(args *ReportTaskArgs, reply *ReportTaskReply) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	switch args.TaskType {
	case MapTask:
		if c.mapTasks[args.TaskId].state != completed {
			c.mapTasks[args.TaskId].state = completed
			c.mapCompleted++
		}
		if c.mapCompleted == c.nMap {
			c.phase = reducePhase
		}
	case ReduceTask:
		if c.reduceTasks[args.TaskId].state != completed {
			c.reduceTasks[args.TaskId].state = completed
			c.reduceCompleted++
		}
		if c.reduceCompleted == c.nReduce {
			c.phase = donePhase
		}
	}
	return nil
}

Whichever report arrives first wins; the second is a no-op. Miss this check and a slow "zombie" worker finishing late can double-count a task, push mapCompleted past nMap, and corrupt the phase transition logic entirely — a bug that would only show up intermittently, exactly when a worker happens to be slow rather than dead, which makes it a nasty one to catch without deliberately testing for it.

Atomic writes: the bug that's easy to skip and hard to debug later

Say a worker is reassigned mid-map-task and still running (it just hasn't heard the news). It's writing to mr-2-3 at the exact moment the reassigned copy also writes to mr-2-3. Without care, a reducer reading that file could see a half-written mess — or one worker's write could clobber the other's mid-write.

The fix is one of the oldest tricks in the book: write somewhere private, then atomically rename into place.

tmpFile, err := ioutil.TempFile(".", "mr-tmp-")
// ... write JSON-encoded key/value pairs to tmpFile ...
tmpFile.Close()
os.Rename(tmpFile.Name(), outname)

os.Rename on the same filesystem is atomic — a reader either sees the file fully present with the old name gone, or doesn't see it yet. There's no in-between state to observe. This is the same pattern databases and log-structured storage engines lean on for crash-safe writes, just applied at the scale of "one text file."

Watching it actually survive a crash

I didn't just trust the design — I built it, then went looking for ways to break it, which is a habit worth having with anything that claims to be fault tolerant.

First, a clean-run correctness check: run the full pipeline on a small set of text files, then diff the concatenated, sorted reduce output against a completely independent ground truth computed with tr/sort/uniq -c:

cat mr-out-* | sort > mr-wc-distributed.txt
cat pg-*.txt | tr -cs 'A-Za-z' '\n' | sort | uniq -c | awk '{print $2, $1}' | sort > mr-wc-groundtruth.txt
diff mr-wc-groundtruth.txt mr-wc-distributed.txt && echo "MATCH"

Exact match. That confirms the happy path — partitioning, shuffling, and reducing all agree with a dumb, obviously-correct sequential computation.

Then the actual test: kill a worker with SIGKILL mid-task, no clean shutdown, no chance for it to tell anyone it's dying, and see if the rest of the swarm carries the job to completion anyway.

./mrworker wc.so &
W1=$!
./mrworker wc.so &
sleep 0.05
kill -9 $W1
./mrworker wc.so &
./mrworker wc.so &
wait $COORD_PID

Job finished. Output matched a hand-computed ground truth exactly, same as the clean run. The ten-second timeout caught the abandoned task, handed it to a live worker, and nothing downstream noticed anything had gone wrong — which is really the entire point of the design. If the ReportTask idempotency check or the atomic-rename logic had a bug, this is exactly the kind of test that would surface it — and it's worth running with a range of sleep delays before the kill, since timing determines whether the task was still idle, freshly inProgress, or already writing files when the worker died.

What this doesn't handle

Worth being honest about the gaps, since a lab-scale implementation and a production one solve different problems:

  • No coordinator fault tolerance. If the coordinator itself dies, the whole job stops — there's no replication, no leader election, no persistent log of task state to recover from. Real MapReduce-style systems (and this lab's own follow-on assignments, like Raft) spend enormous effort solving exactly this for the coordinator's role, not just the workers'.
  • No speculative execution. Real MapReduce famously re-runs tasks that are running slowly (not just dead ones) near the end of a job, since stragglers dominate tail latency at scale. This coordinator only reacts to tasks that have gone completely silent past the timeout.
  • No data locality. The original paper's coordinator tries to schedule map tasks on machines that already have the relevant input data on local disk, since network I/O was the bottleneck at Google's scale. Everything here runs on one machine with a shared filesystem, so locality isn't a consideration.
  • Fixed timeout, no adaptive backoff. Ten seconds is a reasonable constant for a lab-sized job on one machine; it's not something you'd hardcode in a system meant to run heterogeneous tasks across a real cluster.

None of these are bugs — they're scope. Solving all of them is most of what makes systems like this genuinely hard to build for production, and it's a good chunk of what the rest of a course like 6.5840 (Raft, sharded key-value stores) actually gets into.

Final Thoughts

The two things that make MapReduce a distributed systems exercise rather than just "two functions and a for-loop" are the partitioning scheme (turning a shared-state problem — "get every value for a key in one place" — into a naming convention that requires zero coordination between workers) and the timeout-based failure detection (turning "a worker died" into "just give the task to someone else," instead of building anything resembling consensus or heartbeats).

Neither idea is complicated on its own. What's interesting is how little else you need once you have them — no leader election, no distributed locks, no coordination between workers at all. The coordinator does all the bookkeeping and every worker is stateless and disposable, which is exactly the property you want when you assume things will crash.

Full source is on GitHub under mapreduce-go, including the RPC definitions, coordinator, worker, and a word-count app for testing it end to end.