Difficulty: Intermediate-Advanced | Prerequisites: Python sockets, threading basics, JSON, basic OS concepts
You are building a Python MapReduce framework with a Manager process that accepts jobs and distributes work, and multiple Worker processes that execute map and reduce tasks. The Manager and Workers communicate over TCP (messages) and UDP (heartbeats), using JSON-formatted strings. The whole system handles fault tolerance: if a Worker dies mid-task, the Manager detects the missed heartbeats and reassigns that work.
Manager
The central coordinating process. It listens on a TCP port for job submissions and Worker messages, and on a UDP port for heartbeats. It partitions input, assigns tasks, tracks Worker state, and handles fault tolerance.
In simple terms, the Manager is the boss that hands out work and keeps tabs on who is alive.
Worker
A process that registers with the Manager, then executes map or reduce tasks it receives. Each Worker listens on its own TCP port and sends UDP heartbeats every 2 seconds.
Think of it as a single employee that does whatever the Manager tells it to do, one task at a time.
MapReduce job
A complete unit of work submitted by a client. It specifies an input directory, output directory, mapper executable, reducer executable, number of mappers, and number of reducers. Jobs are queued and run one at a time to completion.
Map task (new_map_task)
One partition of the input files, assigned to a single Worker. The Worker runs the mapper executable on each input file, partitions the output by key hash, sorts each partition, and moves the results to the Manager's shared temporary directory.
Reduce task (new_reduce_task)
One partition of the intermediate files from the Map Stage, assigned to a single Worker. The Worker merges the sorted input files using heapq.merge(), pipes them through the reducer executable, and writes the output to the final output directory.
Heartbeat
A UDP message sent by each Worker to the Manager every 2 seconds. If the Manager misses more than 5 consecutive heartbeats from a Worker (10 seconds), it marks that Worker as dead.
In simple terms, it is the Worker saying "I am still alive."
Shared temporary directory
A directory created by the Manager using tempfile.TemporaryDirectory with the prefix "mapreduce-shared-". It holds intermediate files that Workers produce during the Map Stage and that other Workers read during the Reduce Stage.
Partition (in the mapping context)
A subset of map output lines grouped by key hash modulo the number of reducers. This ensures all occurrences of the same key end up at the same reducer.
The framework has three actors: the Client, the Manager, and one or more Workers.
The Client submits a job via mapreduce-submit, which sends a JSON message to the Manager's TCP socket. It plays no further role after submission.
The Manager is a long-running process that:
Creates a shared temporary directory on startup (tempfile.TemporaryDirectory, prefix "mapreduce-shared-")
Starts a UDP listener thread for heartbeats
Starts a TCP listener on the given port for all other messages
Maintains a job queue and runs jobs one at a time, to completion
Tracks every registered Worker and its state (ready, busy, dead)
On shutdown, forwards the shutdown message to all living Workers, then exits
The Manager accepts five command-line options: host, port (shared by TCP and UDP), logfile, loglevel, and shared_dir.
Each Worker is a separate process that:
Opens a TCP listener on its own port
Sends a register message to the Manager
After receiving register_ack, begins sending UDP heartbeats every 2 seconds
Waits for map or reduce tasks from the Manager, executes them one at a time, and sends a finished message back
The Worker accepts six command-line options: host, port, manager-host, manager-port, logfile, loglevel.
All intermediate files live in the Manager's shared temporary directory, under a per-job subdirectory named job-{job_id} (5-digit zero-padded). Each Worker also creates its own local temporary directory (prefix "mapreduce-local-task{task_id}-") for work in progress, then moves sorted results into the shared directory.
All communication is JSON-encoded strings sent over sockets. TCP is used for all command messages (each sent on a fresh connection). UDP is used solely for heartbeats.
The full set of message types is:
register (Worker to Manager, TCP): includes worker_host and worker_port. Sent once the Worker's TCP listener is up.
register_ack (Manager to Worker, TCP): confirms registration. Worker begins heartbeating after receiving this.
new_manager_job (Client to Manager, TCP): contains input_directory, output_directory, mapper_executable, reducer_executable, num_mappers, num_reducers.
new_map_task (Manager to Worker, TCP): contains task_id, input_paths (list), executable, output_directory (the shared temp dir), num_partitions.
new_reduce_task (Manager to Worker, TCP): contains task_id, executable, input_paths (list of intermediate files), output_directory (final output).
finished (Worker to Manager, TCP): contains task_id, worker_host, worker_port. Sent after completing a map or reduce task.
heartbeat (Worker to Manager, UDP): contains worker_host, worker_port. Sent every 2 seconds.
shutdown (Client to Manager, TCP): the Manager forwards it to all living Workers, then both Manager and Workers terminate.
Critical rule: invalid messages (including those that fail JSON decoding) must be silently ignored using a try/except around json.loads().
Workers cycle through three states:
ready: available to accept a new task
busy: currently executing a task
dead: missed more than 5 consecutive heartbeats (10 seconds), or the Manager got a ConnectionRefusedError when trying to send it a TCP message
The registration flow:
Worker starts its TCP listener
Worker sends {"message_type": "register", "worker_host": ..., "worker_port": ...} to the Manager
Manager records the Worker and responds with {"message_type": "register_ack"}
Worker receives register_ack and starts its heartbeat thread (UDP, every 2 seconds)
After the first Worker registers, the Manager checks the job queue for pending work. If a job is already queued or in progress, the Manager assigns the next available task immediately.
The Manager should ignore heartbeats from Workers that have not yet successfully registered. When a Worker re-registers (dies and comes back), all tasks from the running job still need to complete. Mark dead Workers as dead rather than removing them from the Manager's data structures, because modifying a dictionary while iterating over it raises a RuntimeError in Python.
Task assignment order follows registration order. The Manager assigns one task at a time to each Worker, in the order Workers registered.
Students often think the Manager runs multiple jobs concurrently. It does not. Jobs run one at a time, strictly in queue order, each to full completion before the next starts.
Students sometimes remove dead Workers from the Manager's dictionary. This causes RuntimeError if you are iterating over it. Mark them dead but leave them in place.
Students assume heartbeats are sent immediately on startup. They should not be. The Worker must wait for register_ack before starting the heartbeat thread.
Students sometimes store all map output in memory. The spec requires O(1) memory usage during mapping: pipe the executable's stdout line by line using subprocess.Popen, writing each line to the correct partition file as it arrives.
This project teaches the core distributed systems pattern of a coordinator (Manager) and workers, which appears in Hadoop, Spark, and cloud services like AWS EMR and Google Dataproc.
TCP vs UDP choice matters: TCP for reliable command messages, UDP for lightweight, loss-tolerant heartbeats. Expect questions about why each protocol was chosen.
Fault tolerance via heartbeat-based failure detection is a foundational concept in distributed systems. Know the 5-missed-heartbeats threshold and the ConnectionRefusedError shortcut.
The Map Stage's O(1) memory constraint (piping output line by line rather than buffering) is a common exam topic about streaming and memory-efficient processing.
The partition function (MD5 hash of key, converted to int, modulo num_reducers) guarantees all identical keys land at the same reducer. This is fundamental to the correctness of MapReduce.
True or False: The Manager can run two MapReduce jobs at the same time. (False)
Fill in the blank: Workers send heartbeats via ______ every ______ seconds. (UDP, 2)
True or False: A Worker should start sending heartbeats as soon as it starts up. (False, it waits for register_ack)
Fill in the blank: If the Manager misses more than ______ consecutive heartbeats, the Worker is marked dead. (5)
True or False: Each outgoing TCP message should reuse the same connection. (False, each uses a new connection)
Q: Why does the Manager use both TCP and UDP?
A: TCP is used for command messages (register, task assignments, finished, shutdown) because these must be delivered reliably. UDP is used for heartbeats because they are sent frequently, a lost heartbeat is acceptable (the system tolerates up to 5 missed), and UDP avoids the overhead of establishing a connection every 2 seconds.
Q: A Worker dies while executing a map task. Describe what happens next.
A: The Manager detects the death either by missing 5 consecutive heartbeats or by receiving a ConnectionRefusedError when trying to send a message. It marks the Worker as dead (but does not remove it from its data structures). The unfinished task is placed back in the queue. Once a ready Worker is available, the Manager reassigns that task. The Manager finishes assigning all normal tasks in the current stage before reassigning tasks from dead Workers.
Q: How does the partition function ensure correctness in MapReduce?
A: The partition function hashes each key with MD5, converts the hex digest to an integer, and takes modulo num_reducers. Because the same key always produces the same hash, all lines with a given key land in the same partition file and therefore reach the same reducer. This is essential for correct aggregation.
Q: Why does the Worker create a local temporary directory rather than writing directly to the Manager's shared directory?
A: In a real distributed deployment, the Worker's local disk is faster than writing over the network to a shared file system. The Worker writes and sorts locally, then moves the finished files to the shared directory. This pattern minimises network I/O.
Q: What happens if a job is submitted before any Workers have registered?
A: The Manager adds the job to its queue. When the first Worker registers, the Manager checks the queue and begins assigning tasks immediately.
MapReduce, distributed systems, Manager-Worker architecture, coordinator pattern, fault tolerance, heartbeat protocol, TCP sockets, UDP sockets, Python threading, subprocess.Popen, tempfile.TemporaryDirectory, heapq.merge, hash partitioning, round-robin partitioning, EECS 485 Project 4, p4-mapreduce, Google MapReduce paper, Hadoop streaming, Madoop, AWS EMR, Google Dataproc