Difficulty: Intermediate-Advanced | Prerequisites: Part 1 study notes (Overview), subprocess, hashlib, heapq
A MapReduce job flows through three stages: Map (run mapper on input partitions, hash-partition and sort the output), Group (the Manager gathers intermediate files by partition number), and Reduce (merge sorted inputs, run reducer, write final output). The Worker does the compute-heavy work; the Manager orchestrates file assignments and stage transitions.
Input partitioning (round robin)
The Manager sorts input files alphabetically, then distributes them across num_mappers partitions using round robin. Partition 0 gets files at indices 0, num_mappers, 2*num_mappers, and so on.
Hash partitioning
During the Map Stage, each output line's key is hashed with MD5, converted to an integer (base 16), and taken modulo num_partitions (which equals num_reducers). This determines which intermediate file the line is written to.
Intermediate files
Files named maptask{task_id}-part{partition_number}, both zero-padded to 5 digits. They live in the Manager's shared temporary directory after the Worker moves them there.
heapq.merge()
A Python standard library function that merges multiple sorted iterables into a single sorted iterator without loading everything into memory. Used in the Reduce Stage to merge sorted intermediate files.
subprocess.Popen
Used to run map and reduce executables as child processes. For mapping, stdin comes from a file and stdout is piped. For reducing, stdin is piped (from heapq.merge output) and stdout goes to a file.
When a job begins, the Manager scans the input directory, sorts file names alphabetically, and divides them into num_mappers partitions using round robin.
Worked example with 5 files and num_mappers = 4:
Sorted files: file01, file02, file03, file04, file05
Partition 0 (task_id 0): file01, file05
Partition 1 (task_id 1): file02
Partition 2 (task_id 2): file03
Partition 3 (task_id 3): file04
Each partition becomes a new_map_task. The task_id starts at 0 and increments. Tasks are assigned to Workers in the order Workers registered, one at a time. If all Workers are busy, the Manager waits until one finishes.
The output_directory in the new_map_task message is the Manager's shared temporary directory (inside the job-{job_id} subdirectory). num_partitions equals num_reducers.
The Worker performs four steps for each map task:
Step 1: Run the mapper executable. For each input file, open it and pipe it as stdin to the mapper using subprocess.Popen. The mapper's stdout is piped back to the Worker (stdout=subprocess.PIPE). Process the output line by line as it arrives. This is the O(1) memory requirement.
with open(input_path) as infile:
with subprocess.Popen([executable], stdin=infile,
stdout=subprocess.PIPE, text=True) as proc:
for line in proc.stdout:
# write line to correct partition fileStep 2: Partition the output. Each output line is a tab-separated key-value pair. Extract the key (everything before the first tab), hash it, and write the line to the corresponding partition file.
import hashlib
hexdigest = hashlib.md5(key.encode("utf-8")).hexdigest()
keyhash = int(hexdigest, base=16)
partition_number = keyhash % num_partitionsIntermediate files are named maptask{task_id}-part{partition_number} (both 5-digit zero-padded) and written inside a Worker-local temporary directory (prefix "mapreduce-local-task{task_id}-").
Step 3: Sort each partition file. After all input files are processed, sort each intermediate file in place using UNIX sort via subprocess:
subprocess.run(["sort", "-o", filename, filename], check=True)Step 4: Move sorted files to the shared directory. Move every sorted intermediate file from the Worker's local temporary directory into the output_directory specified in the new_map_task message (the Manager's shared temp directory).
After completing all four steps, the Worker sends a finished message back to the Manager.
Once all map tasks are complete, the Manager transitions to the Reduce Stage. It groups the intermediate files in the shared temporary directory by partition number.
For example, if two map tasks produced these files:
maptask00000-part00000
maptask00000-part00001
maptask00001-part00000
maptask00001-part00001
The Manager creates two reduce tasks:
task_id 0 gets input_paths: [maptask00000-part00000, maptask00001-part00000]
task_id 1 gets input_paths: [maptask00000-part00001, maptask00001-part00001]
Use pathlib's glob to collect files matching each partition. The output_directory in the new_reduce_task message is the final output directory from the original job submission. Tasks are assigned to Workers in registration order, just as in the Map Stage.
The Worker performs three steps for each reduce task:
Step 1: Merge input files. Each input file is already sorted from the Map Stage. Open all input files and merge them into a single sorted stream using heapq.merge(). Do not load everything into memory; heapq.merge returns a lazy iterator.
Step 2: Run the reducer executable. Pipe the merged stream into the reducer's stdin using subprocess.Popen. The reducer's stdout goes to an output file named part-{task_id} (5-digit zero-padded), written inside a Worker-local temporary directory (prefix "mapreduce-local-task{task_id}-").
executable = # reduce executable
instream = # heapq.merge(*open_files)
outfile = # open output file
with subprocess.Popen([executable], text=True,
stdin=subprocess.PIPE,
stdout=outfile) as reduce_process:
for line in instream:
reduce_process.stdin.write(line)Note: use contextlib.ExitStack to open multiple files at once while satisfying pylint's context-manager requirement.
Step 3: Move the output file. Move part-{task_id} from the local temporary directory to the output_directory specified in the new_reduce_task message (the final output directory from the original job).
After completing all three steps, the Worker sends a finished message back to the Manager.
After all reduce tasks finish, the Manager deletes the job's intermediate directory (e.g. job-00000) from the shared temporary directory. This cleanup also happens automatically on shutdown via the tempfile context manager, but must happen explicitly at the end of each job as well.
The Manager then checks its queue. If another job is waiting, it pops it off and begins execution immediately.
Students often think the Map Stage can buffer all output lines in a list before partitioning. The spec requires O(1) memory: read from subprocess.Popen's stdout line by line and write each line to its partition file immediately.
Students sometimes confuse num_mappers with num_partitions. num_mappers controls how many map tasks the input is split into. num_partitions (in the map task message) equals num_reducers and controls how many partition files each mapper produces.
Students forget that the reduce input must be globally sorted across all input files. You cannot simply concatenate the files; you must use heapq.merge to produce a single sorted stream.
Students occasionally create the output file directly in the final output directory. The correct pattern is to write to a Worker-local temp directory first, then move the file. This mirrors how a real distributed system minimises network writes.
The hash-partition-then-reduce pattern is how every MapReduce system guarantees that all values for a given key reach the same reducer. Know the formula: MD5 hash, int(hex, 16), modulo num_reducers.
O(1) memory in the Map Stage is a deliberate design choice. Exam questions may ask why you pipe output rather than collecting it, or what happens if input is larger than available memory.
The merge step in the Reduce Stage is a classic k-way merge problem. heapq.merge handles it efficiently with a min-heap, maintaining O(k) memory where k is the number of input files.
File naming conventions (maptask{task_id}-part{partition_number}, part-{task_id}, all 5-digit zero-padded) are load-bearing for the autograder. Deviations will cause test failures.
True or False: num_partitions in the map task message equals num_mappers. (False, it equals num_reducers)
Fill in the blank: Intermediate files are named maptask{}-part{}, both zero-padded to _____ digits. (task_id, partition_number, 5)
True or False: The reducer's input files can be concatenated in any order. (False, they must be merged in sorted order using heapq.merge)
Fill in the blank: The Worker writes output to a _____ directory first, then moves it to the _____. (local temporary, shared/output directory)
True or False: The Map Stage should use O(n) memory where n is the size of the map output. (False, O(1))
Q: Walk through what happens to the word "hello" appearing in two different input files processed by two different map tasks, with num_reducers = 2.
A: Each mapper outputs "hello\t1" for every occurrence. The key "hello" is hashed with MD5 and taken modulo 2, which always yields the same partition number (say, partition 0). Both mappers write their "hello\t1" lines to maptask{their_task_id}-part00000. These files are sorted. During the Reduce Stage, the Manager groups both part00000 files and sends them to the reducer with task_id 0. The reducer merges them (already sorted), sees all "hello" lines together, sums the counts, and outputs "hello\t3" (if hello appeared 3 times total).
Q: Why does the Worker sort each intermediate file after writing it?
A: Sorting at the map output stage means the Reduce Stage can use a k-way merge (heapq.merge) rather than a full sort. k-way merging sorted streams is O(n log k) and uses only O(k) memory, whereas sorting an unsorted file of n lines would be O(n log n) and might not fit in memory.
Q: What happens if you use subprocess.run instead of subprocess.Popen for the map executable?
A: subprocess.run waits for the process to complete and returns all of stdout at once. This means the entire map output is buffered in memory, violating the O(1) memory requirement. subprocess.Popen lets you iterate over stdout line by line as it is produced.
Q: Suppose num_mappers = 3 and you have 7 input files (file01 through file07, sorted). What are the partitions?
A: Round-robin assignment: partition 0 gets file01, file04, file07; partition 1 gets file02, file05; partition 2 gets file03, file06.
Map Stage, Reduce Stage, Group Stage, input partitioning, round robin, hash partition, MD5 hash, modulo, subprocess.Popen, subprocess.run, heapq.merge, k-way merge, UNIX sort, intermediate files, maptask, part file, O(1) memory, streaming, pipe, stdin, stdout, tempfile.TemporaryDirectory, contextlib.ExitStack, pathlib.glob, shutil.move, EECS 485 P4 MapReduce execution