p4-mapreduce
Walk-through example
This document walks step-by-step through an example MapReduce job. Our staff Manager and Worker solutions produce logging output so you can see what’s going on. For brevity, we’ve omitted heartbeat messages and the sender’s copy of each TCP message.
Here’s an overview of the messages between the Client, Manager and Worker.
Files
Start in your project root directory. Take a look at a few of the input files.
$ pwd
/Users/awdeorio/src/eecs485/p4-mapreduce
$ ls tests/testdata/input_small
file01 file02
$ cat tests/testdata/input_small/file01
hello world
hello eecs485
$ cat tests/testdata/input_small/file02
goodbye autograder
hello autograder
We’re going to run a word count job. Start by simulating the MapReduce job at the command line. This is the expected output. Save it to a file.
$ cat tests/testdata/input_small/* | tests/testdata/exec/wc_map.sh | sort | tests/testdata/exec/wc_reduce.sh
autograder 2
eecs485 1
goodbye 1
hello 3
world 1
$ cat tests/testdata/input_small/* | tests/testdata/exec/wc_map.sh | sort | tests/testdata/exec/wc_reduce.sh > correct.txt
These are the files we have before starting the server.
$ pwd
/Users/awdeorio/src/eecs485/p4-mapreduce
$ ls
bin mapreduce mapreduce.egg-info requirements.txt pyproject.toml tests
Start Manager
Kill any stale MapReduce processes.
$ pkill -lf mapreduce- # macOS
$ pkill -f mapreduce- # Linux/WSL
Start the Manager.
$ mapreduce-manager --loglevel=DEBUG
Manager:6000 [INFO] Manager host=localhost port=6000 pwd=/Users/awdeorio/src/eecs485/p4-mapreduce
Manager:6000 [INFO] Created /tmp/mapreduce-shared-7r6il3a5
Manager:6000 [INFO] Start UDP server thread
Manager:6000 [INFO] Start TCP server thread
Manager:6000 [DEBUG] TCP bind localhost:6000
Manager:6000 [DEBUG] UDP bind localhost:6000
The Manager created a shared temporary directory /tmp/mapreduce-shared-7r6il3a5. We’re using the Python standard library tempfile.TemporaryDirectory to create and automatically clean up the temporary directory, so your path could be different.
Start Workers
Start two Workers in separate terminals.
$ mapreduce-worker --port 6001 --loglevel=DEBUG # Terminal 2
Worker:6001 [INFO] Worker host=localhost port=6001 manager_host=localhost, manager_port=6000 pwd=/Users/awdeorio/src/eecs485/p4-mapreduce
Worker:6001 [INFO] Start TCP server thread
Worker:6001 [DEBUG] TCP bind localhost:6001
Worker:6001 [INFO] Sent connection request to Manager localhost:6000
Manager:6000 [DEBUG] TCP recv
{
"message_type": "register",
"worker_host": "localhost",
"worker_port": 6001
}
Manager:6000 [INFO] Registered Worker RemoteWorker('localhost', 6001)
Worker:6001 [DEBUG] TCP recv
{
"message_type": "register_ack"
}
Worker:6001 [DEBUG] Got register_ack {'message_type': 'register_ack'}
Worker:6001 [INFO] Connected to Manager localhost:6000
Worker:6001 [INFO] Start heartbeat thread
$ mapreduce-worker --port 6002 --loglevel=DEBUG # Terminal 3
Worker:6002 [INFO] Worker host=localhost port=6002 manager_host=localhost, manager_port=6000 pwd=/Users/awdeorio/src/eecs485/p4-mapreduce
Worker:6002 [INFO] Start TCP server thread
Worker:6002 [DEBUG] TCP bind localhost:6002
Worker:6002 [INFO] Sent connection request to Manager localhost:6000
Manager:6000 [DEBUG] TCP recv
{
"message_type": "register",
"worker_host": "localhost",
"worker_port": 6002
}
Manager:6000 [INFO] Registered Worker RemoteWorker('localhost', 6002)
Worker:6002 [DEBUG] TCP recv
{
"message_type": "register_ack"
}
Worker:6002 [DEBUG] Got register_ack {'message_type': 'register_ack'}
Worker:6002 [INFO] Connected to Manager localhost:6000
Worker:6002 [INFO] Start heartbeat thread
$
At this point, you’ll have 3 terminals open for the Manager and two Workers.
Pro-tip To produce logging output similar to the instructor solution, see the Logging section of the spec.
Submit a MapReduce job
Submit a word count job in a fourth terminal. We’re taking advantage of the submit defaults, see mapreduce-submit --help for more info.
$ mapreduce-submit -i tests/testdata/input_small
Submitted job to Manager localhost:6000
input directory tests/testdata/input_small
output directory output
mapper executable tests/testdata/exec/wc_map.sh
reducer executable tests/testdata/exec/wc_reduce.sh
num mappers 2
num reducers 2
The Manager receives the job.
Manager:6000 [DEBUG] TCP recv
{
"message_type": "new_manager_job",
"input_directory": "tests/testdata/input_small",
"output_directory": "output",
"mapper_executable": "tests/testdata/exec/wc_map.sh",
"reducer_executable": "tests/testdata/exec/wc_reduce.sh",
"num_mappers": 2,
"num_reducers": 2
}
Manager:6000 [INFO] Received job Job(job_id=0, finished=False)
Manager:6000 [INFO] Created output
Manager:6000 [INFO] Created /tmp/mapreduce-shared-7r6il3a5/job-00000
The Manager created a final output directory output and a job-00000 directory for intermediate files inside of the shared temporary directory.
Map Stage
The Map Stage runs.
Manager:6000 [INFO] Begin Map Stage
Manager:6000 [DEBUG] stage=Stage.MAP prev_stage=Stage.CREATED
Manager:6000 [INFO] Assigned task to Worker Task(task_id=0, input_paths=['file01'], executable='wc_map.sh', output_directory='job-00000', num_partitions=2) RemoteWorker('localhost', 6001)
Worker:6001 [DEBUG] TCP recv
{
"message_type": "new_map_task",
"task_id": 0,
"input_paths": [
"tests/testdata/input_small/file01"
],
"executable": "tests/testdata/exec/wc_map.sh",
"output_directory": "/tmp/mapreduce-shared-7r6il3a5/job-00000",
"num_partitions": 2
}
Worker:6001 [INFO] Received task Task(task_id=0, input_paths=['file01'], executable='wc_map.sh', output_directory='job-00000', num_partitions=2)
Worker:6001 [INFO] Created /tmp/mapreduce-local-task00000-umi_e5q7
Worker:6001 [INFO] Executed tests/testdata/exec/wc_map.sh input=tests/testdata/input_small/file01
Worker:6001 [INFO] Sorted /tmp/mapreduce-local-task00000-umi_e5q7/maptask00000-part00000
Worker:6001 [INFO] Sorted /tmp/mapreduce-local-task00000-umi_e5q7/maptask00000-part00001
Worker:6001 [INFO] Moved /tmp/mapreduce-local-task00000-umi_e5q7/maptask00000-part00000 -> /tmp/mapreduce-shared-7r6il3a5/job-00000/maptask00000-part00000
Worker:6001 [INFO] Moved /tmp/mapreduce-local-task00000-umi_e5q7/maptask00000-part00001 -> /tmp/mapreduce-shared-7r6il3a5/job-00000/maptask00000-part00001
Worker:6001 [INFO] Removed /tmp/mapreduce-local-task00000-umi_e5q7
Worker:6001 [INFO] Completed task Task(task_id=0, input_paths=['file01'], executable='wc_map.sh', output_directory='job-00000', num_partitions=2)
Manager:6000 [DEBUG] TCP recv
{
"message_type": "finished",
"task_id": 0,
"worker_host": "localhost",
"worker_port": 6001
}
Manager:6000 [INFO] Finished task from Worker Task(task_id=0, input_paths=['file01'], executable='wc_map.sh', output_directory='job-00000', num_partitions=2) RemoteWorker('localhost', 6001)
Manager:6000 [DEBUG] stage=Stage.MAP prev_stage=Stage.MAP
Manager:6000 [INFO] Assigned task to Worker Task(task_id=1, input_paths=['file02'], executable='wc_map.sh', output_directory='job-00000', num_partitions=2) RemoteWorker('localhost', 6001)
Worker:6001 [DEBUG] TCP recv
{
"message_type": "new_map_task",
"task_id": 1,
"input_paths": [
"tests/testdata/input_small/file02"
],
"executable": "tests/testdata/exec/wc_map.sh",
"output_directory": "/tmp/mapreduce-shared-7r6il3a5/job-00000",
"num_partitions": 2
}
Worker:6001 [INFO] Received task Task(task_id=1, input_paths=['file02'], executable='wc_map.sh', output_directory='job-00000', num_partitions=2)
Worker:6001 [INFO] Created /tmp/mapreduce-local-task00001-l84s1em6
Worker:6001 [INFO] Executed tests/testdata/exec/wc_map.sh input=tests/testdata/input_small/file02
Worker:6001 [INFO] Sorted /tmp/mapreduce-local-task00001-l84s1em6/maptask00001-part00000
Worker:6001 [INFO] Sorted /tmp/mapreduce-local-task00001-l84s1em6/maptask00001-part00001
Worker:6001 [INFO] Moved /tmp/mapreduce-local-task00001-l84s1em6/maptask00001-part00000 -> /tmp/mapreduce-shared-7r6il3a5/job-00000/maptask00001-part00000
Worker:6001 [INFO] Moved /tmp/mapreduce-local-task00001-l84s1em6/maptask00001-part00001 -> /tmp/mapreduce-shared-7r6il3a5/job-00000/maptask00001-part00001
Worker:6001 [INFO] Removed /tmp/mapreduce-local-task00001-l84s1em6
Worker:6001 [INFO] Completed task Task(task_id=1, input_paths=['file02'], executable='wc_map.sh', output_directory='job-00000', num_partitions=2)
Manager:6000 [DEBUG] TCP recv
{
"message_type": "finished",
"task_id": 1,
"worker_host": "localhost",
"worker_port": 6001
}
Manager:6000 [INFO] Finished task from Worker Task(task_id=1, input_paths=['file02'], executable='wc_map.sh', output_directory='job-00000', num_partitions=2) RemoteWorker('localhost', 6001)
Manager:6000 [INFO] End Map Stage
In this run, Worker 6001 finished each task before the Manager assigned the next one, so Worker 6001 ran every task. Your Manager may split tasks between Workers differently.
Note how the Workers created local temporary directories of their own for each task, and then moved the output files into the Manager’s shared temporary directory when they finished writing and sorting.
At the end of the Map Stage, mapping, partitioning, and sorting is complete. In this example, the Manager’s shared temporary directory (e.g., /tmp/mapreduce-shared-XYZ123/job-00000) contains 4 files. To see these files during debugging, set a breakpoint before the Manager’s shared temporary directory is deleted.
hello 1
hello 1
eecs485 1
world 1
hello 1
autograder 1
autograder 1
goodbye 1
With 2 reducers, we have 2 partitions. For example when map task 0 computes md5("hello") % 2 == 0, this key goes in partition 0, maptask00000-part00000. Similarly, when map task 1 computes md5("hello") % 2 == 0, this key goes in partition 0, maptask00001-part00000.
Reduce Stage
Finally, the Reduce Stage runs.
Manager:6000 [INFO] Begin Reduce Stage
Manager:6000 [DEBUG] stage=Stage.REDUCE prev_stage=Stage.MAP
Manager:6000 [INFO] Assigned task to Worker Task(task_id=0, input_paths=['maptask00000-part00000', 'maptask00001-part00000'], executable='wc_reduce.sh', output_directory='output') RemoteWorker('localhost', 6001)
Worker:6001 [DEBUG] TCP recv
{
"message_type": "new_reduce_task",
"task_id": 0,
"input_paths": [
"/tmp/mapreduce-shared-7r6il3a5/job-00000/maptask00000-part00000",
"/tmp/mapreduce-shared-7r6il3a5/job-00000/maptask00001-part00000"
],
"executable": "tests/testdata/exec/wc_reduce.sh",
"output_directory": "output"
}
Worker:6001 [INFO] Received task Task(task_id=0, input_paths=['maptask00000-part00000', 'maptask00001-part00000'], executable='wc_reduce.sh', output_directory='output')
Worker:6001 [INFO] Created /tmp/mapreduce-local-task00000-jovdb59e
Worker:6001 [INFO] Executed tests/testdata/exec/wc_reduce.sh
Worker:6001 [INFO] Moved /tmp/mapreduce-local-task00000-jovdb59e/part-00000 -> output/part-00000
Worker:6001 [INFO] Removed /tmp/mapreduce-local-task00000-jovdb59e
Worker:6001 [INFO] Completed task Task(task_id=0, input_paths=['maptask00000-part00000', 'maptask00001-part00000'], executable='wc_reduce.sh', output_directory='output')
Manager:6000 [DEBUG] TCP recv
{
"message_type": "finished",
"task_id": 0,
"worker_host": "localhost",
"worker_port": 6001
}
Manager:6000 [INFO] Finished task from Worker Task(task_id=0, input_paths=['maptask00000-part00000', 'maptask00001-part00000'], executable='wc_reduce.sh', output_directory='output') RemoteWorker('localhost', 6001)
Manager:6000 [DEBUG] stage=Stage.REDUCE prev_stage=Stage.REDUCE
Manager:6000 [INFO] Assigned task to Worker Task(task_id=1, input_paths=['maptask00000-part00001', 'maptask00001-part00001'], executable='wc_reduce.sh', output_directory='output') RemoteWorker('localhost', 6001)
Worker:6001 [DEBUG] TCP recv
{
"message_type": "new_reduce_task",
"task_id": 1,
"input_paths": [
"/tmp/mapreduce-shared-7r6il3a5/job-00000/maptask00000-part00001",
"/tmp/mapreduce-shared-7r6il3a5/job-00000/maptask00001-part00001"
],
"executable": "tests/testdata/exec/wc_reduce.sh",
"output_directory": "output"
}
Worker:6001 [INFO] Received task Task(task_id=1, input_paths=['maptask00000-part00001', 'maptask00001-part00001'], executable='wc_reduce.sh', output_directory='output')
Worker:6001 [INFO] Created /tmp/mapreduce-local-task00001-eghs47n6
Worker:6001 [INFO] Executed tests/testdata/exec/wc_reduce.sh
Worker:6001 [INFO] Moved /tmp/mapreduce-local-task00001-eghs47n6/part-00001 -> output/part-00001
Worker:6001 [INFO] Removed /tmp/mapreduce-local-task00001-eghs47n6
Worker:6001 [INFO] Completed task Task(task_id=1, input_paths=['maptask00000-part00001', 'maptask00001-part00001'], executable='wc_reduce.sh', output_directory='output')
Manager:6000 [DEBUG] TCP recv
{
"message_type": "finished",
"task_id": 1,
"worker_host": "localhost",
"worker_port": 6001
}
Manager:6000 [INFO] Finished task from Worker Task(task_id=1, input_paths=['maptask00000-part00001', 'maptask00001-part00001'], executable='wc_reduce.sh', output_directory='output') RemoteWorker('localhost', 6001)
Manager:6000 [INFO] End Reduce Stage
Manager:6000 [DEBUG] stage=Stage.FINISHED prev_stage=Stage.REDUCE
Manager:6000 [INFO] Removed /tmp/mapreduce-shared-7r6il3a5/job-00000
Manager:6000 [INFO] Finished job Job(job_id=0, finished=True)
Output
After the job is finished, we can see the final output inside output/.
$ tree output/
output/
├── part-00000
└── part-00001
hello 3
autograder 2
eecs485 1
goodbye 1
world 1
Recall that md5("hello") % 2 == 0, so this key is in partition 0 (part00000). Similarly, md5("autograder") % 2 == 1, so this key is in partition 1 (part00001).
Sort the output so that we can easily compare with the correct answer.
$ cat output/* | sort
autograder 2
eecs485 1
goodbye 1
hello 3
world 1
$ cat output/* | sort > actual.txt
$ diff correct.txt actual.txt
Shutdown
Finally, send a shutdown signal to the server.
$ mapreduce-submit --shutdown # Terminal 4
Manager:6000 [DEBUG] TCP recv
{
"message_type": "shutdown"
}
Manager:6000 [INFO] Received shutdown {'message_type': 'shutdown'}
Manager:6000 [INFO] Sent shutdown to Worker RemoteWorker('localhost', 6001)
Worker:6001 [DEBUG] TCP recv
{
"message_type": "shutdown"
}
Worker:6001 [INFO] Received shutdown {'message_type': 'shutdown'}
Worker:6001 [INFO] Shutdown TCP server thread
Manager:6000 [INFO] Sent shutdown to Worker RemoteWorker('localhost', 6002)
Manager:6000 [INFO] Shutdown TCP server thread
Worker:6002 [DEBUG] TCP recv
{
"message_type": "shutdown"
}
Worker:6002 [INFO] Received shutdown {'message_type': 'shutdown'}
Worker:6002 [INFO] Shutdown TCP server thread
Manager:6000 [INFO] Removed /tmp/mapreduce-shared-7r6il3a5
Manager:6000 [INFO] Shutdown UDP server thread
Manager:6000 [INFO] Shutdown main thread
Worker:6002 [INFO] Shutdown heartbeat thread
Worker:6002 [INFO] Shutdown main thread
Worker:6001 [INFO] Shutdown heartbeat thread
Worker:6001 [INFO] Shutdown main thread
After you run the shutdown command and get the above output, hit Enter.
You should see something similar to the following:
[1] Done mapreduce-manager
[2]- Done mapreduce-worker
[3]+ Done mapreduce-worker --port 6002
Verify that no MapReduce processes are running.
$ pgrep -lf mapreduce- # macOS
$ pgrep -af mapreduce- # Linux
Acknowledgments
This document is licensed under a Creative Commons Attribution-NonCommercial 4.0 License. You’re free to copy and share this document, but not to sell it. You may not share source code provided with this document.