EECS 485 Project 4: Map Reduce

Due: 8pm on March 27th, 2019. This is a group project to be completed in groups of two to three.

Change Log


In this project, you will implement a MapReduce server in Python. This will be a single machine, multi-process, multi-threaded server that will execute user-submitted MapReduce jobs. It will run each job to completion, handling failures along the way, and write the output of the job to a given directory. Once you have completed this project, you will be able to run any MapReduce job on your machine, using a MapReduce implementation you wrote!

There are two primary modules in this project: the Master , which will listen for MapReduce jobs, manage the jobs, distribute work amongst the workers, and handle faults. Worker modules register themselves with the master, and then await commands, performing map, reduce or sorting (grouping) tasks based on instructions given by the master.

You will not write map reduce programs, but rather the MapReduce server. We have provided several sample map/reduce programs that you can use to test your MapReduce server.

Refer to the P1 setup tutorial for setting up your development environment.

Refer to the Python processes, threads and sockets tutorial for background and examples.

Table of Contents


Project Structure

You will write a mapreduce Python package includes master and worker modules. Launch a master with the command line entry point mapreduce-master and a worker with mapreduce-worker . We’ve also provided mapreduce-submit to send a new job to the master.

Starter files

We will start with a summary of the starter files.

· mapreduce/master/ : Implement the MapReduce Master here.

· mapreduce/worker/ : Implement the MapReduce Worker here.

· mapreduce/utils.py : Common code shared by Master and Worker (optional).

· tests/ : Public unit tests.

· tests/input/ : Sample input files.

· tests/exec/ : MapReduce programs. All use stdin and stdout.

Download the starter files and copy the contents into your project root directory.


Your code will go inside the mapreduce/master and mapreduce/worker packages, where you will define the two classes (we got you started in mapreduce/master/  main  .py and  mapreduce/worker/ main .py ). Since we are using Python packages, you may create new files as you see fit inside each package. We have also provided a utils.py inside mapreduce/ which you can use to house code common to both Worker and Master. We will only define the communication specs for the Master and the Worker, but the actual implementation of the classes is entirely up to you.

The starter code will run out of the box, it just won’t do anything. At this point, you will be able to pass test_master_0 and test_worker_0 . The master and the worker run as seperate processes, so you will have to start them up separately. This will start up a master which will listen on port 6000 using TCP. Then, we start up two workers, and tell them that they should communicate with the master on port 6000, and then tell them which port to listen on. The ampersand ( & ) means to start the process in the background.


See your processes running in the background. Note: use pgrep -lf on OSX and pgrep -af on GNU/Linux systems.


Lastly, we have also provided mapreduce/submit.py . It sends a job to the Master’s main TCP socket. You can specify the job using command line arguments.


Here’s how to run a job. Later, we’ll simplify starting the server using a bash script. Right now we expect the job to fail because Master and Worker are not implemented.


Init script

The MapReduce server is an example of a service (or daemon), a program that runs in the background. We’ll write an init script to start, stop and check on the map reduce master and worker processes. It should be a bash script named bin/mapreduce . Print the messages in the following examples.

Be sure to follow the bash script best practices https://eecs485staff.github.io/p1-insta485- static/#utility-scripts, e.g., starting with set -Eeuo pipefail .

Start server

Exit 1 if a master or worker is already running. Otherwise, execute the following commands.


NOTE: On the autograder, the “test_scripts” will run with your Master and Worker.

Master Class

The Master should accept only one command line argument. port_number : The primary TCP port that the Master should listen on. On startup, the Master should do the following:

· Create a new folder tmp . This is where we will store all intermediate files used by the MapReduce server.
      If tmp already exists, keep it

      Delete any old mapreduce job folders in tmp . HINT: see the Python glob                  module and use”tmp/job-*” .

· Create a new thread, which will listen for UDP heartbeat messages from the workers. This should listen on ( port_number – 1 )

· Create any additional threads or setup you think you may need. Another thread for fault tolerance could be helpful.

· Create a new TCP socket on the given port_number and call the listen() function. · Wait for incoming messages!

Worker Class

The Worker should accept two command line arguments.

master_port : The TCP socket that the Master is actively listening on (same as the port_number in the Master constructor)

worker_port : The TCP socket that this worker should listen on to receive instructions from the master

On initialization, each Worker should do a similar sequence of actions as the Master:

· Get the process ID of the Worker. This will be the Worker’s unique ID, which it should then use to register with the master.

· Create a new TCP socket on the given worker_port and call the listen() function. Send the register message to the Master

· Upon receiving the register_ack message, create a new thread which will be responsible for sending heartbeat messages to the Master.

NOTE: The Master should safely ignore any heartbeat messages from a Worker before that Worker successfully registers with the Master.


These are some of the libraries that we used in our implementation. We strongly recommend you use these – they will save you an incredible amount of time, and code!

· Python Multithreading

· Python Sockets

· Python SH Module

· Python JSON Library
We have provided examples of threads and sockets in our Python processes, threads and sockets tutorial.

You will use UDP for heartbeat messages and TCP for all other communications. In Python, you can specify the maximum queue size to a socket so that messages aren’t ignored if you’re busy (look at the argument for the listen() function when you get to it). We highly recommend reading through Python TCP and UDP communication documentation.


The Python logging facility is helpful for monitoring multiple processes. For logging output similar to the instructor solution in the walk-through example, try this.


MapReduce Server Specification

Here, we describe the functionality of the MapReduce server. The fun part is that we are only defining the functionality and the communication spec, the implementation is entirely up to you. You must follow our exact specifications below, and the Master and the Worker should work independently (i.e. do not add any more data or dependencies between the two classes). Remember that the Master/Workers are listening on TCP/UDP sockets for all incoming messages. Note: To test your server, we will test your worker with our Master and your Master with our Worker. You should not rely on any communication other than the messages listed below.

As soon as the Master/Worker receives a message on its main TCP socket, it should handle that message to completion before continuing to listen on the TCP socket. In this spec, let’s say every message is handled in a function called handle_msg . When the message returns and ends execution, the Master will continue listening in an infinite while loop for new messages. Each TCP message should be communicated using a new TCP connection. Note: All communication in this project will be strings formatted using JSON; sockets receive strings but your thread must parse it into JSON.

We put [Master/Worker] before the subsections below to identify which class should handle the given functionality.

Worker Registration – [Master + Worker]

The Master should keep track of all Workers at any given time so that the work is only distributed among the ready Workers. Workers can be in the following states:

· ready : Worker is ready to accept work

· busy : Worker is performing a job

· dead : Worker has failed to ping for some amount of time

The Master must listen for registration messages from Workers. Once a Worker is ready to listen for instructions, it should send a message like this to the Master


After the first Worker registers with the Master, the Master should check the job queue (described later) if it has any work it can assign to the Worker (because a job could have arrived at the Master before any Workers registered). If the Master is already executing a map/group/reduce, it can wait until the next phase or wait until completion of the complete current task to assign the Worker any tasks.

At this point, you should be able to pass test_master_1 and test_worker_1 .

New Job Request – [Master]

In the event of a new job, the Master will receive the following message on its main TCP socket:


In response to a job request, the Master will create a set of new directories where all of the temporary files for the job will go, of the form tmp/job-{id} , where id is the current job counter (starting at 0 just like all counters). The directory structure will resemble this example (you should create 4 new folders for each job):


Remember, each MapReduce job occurs in 3 phases: mapping, grouping, reducing. Workers will do the mapping and reducing using the given executable files independently, but the Master and Workers will have to cooperate to do the grouping phase. After the directories are setup, the Master should check if there are any Workers ready to work and check whether the MapReduce server is currently executing a job. If the server is busy, or there are no available Workers, the job should be added to an internal queue (described next) and end the function execution. If there are workers and the server is not busy, than the Master can begin job execution.
At this point, you should be able to pass test_master_2 .

Job Queue – [Master]
If a Master receives a new job while it is already executing one or when there were no ready workers, it should accept the job, create the directories, and store the job in an internal queue until the current one has finished. Note that this means that the current job’s map, group, and reduce tasks must be complete before the next job’s map phase can begin. As soon as a job finishes, the Master should process the next pending job if there is one (and if there are ready Workers) by starting its Map stage. For simplicity, in this project, your MapReduce server will only execute one MapReduce task at any time.

As noted earlier, when you see the first Worker register to work, you should check the job queue for pending jobs.

Input Partitioning – [Master]

To start off the Map Stage, the Master should scan the input directory and partition the input files in ‘X’ parts (where ‘X’ is the number of map tasks specified in the incoming job). After partitioning the input, the Master needs to let each Worker know what work it is responsible for. Each Worker could get zero, one, or many such tasks. The Master will send a JSON message of the following form to each Worker (on each Worker’s specific TCP socket), letting them know that they have work to do:


Consider the case where there are 2 Workers available, 5 input files and 4 map tasks specified. The master should create 4 tasks, 3 with one input file each and 1 with 2 input files. It would then attempt to balance these tasks among all the workers. In this case, it would send 2 map tasks to each worker. The master does not need to wait for a done message before it assigns more tasks to a Worker – a Worker should be able to handle multiple tasks at the same time.

Mapping – [Workers]

When a worker receives this new job message, its handle_msg will start execution of the given executable over the specified input file, while directing the output to the given output_directory (one output file per input file and you should run the executable on each input file). The input is passed to the executable through standard in and is outputted to a specific file. The output file names should be the same as the input file (overwrite file if it already exists). The output_directory in the Map stage will always be the mapper-output folder (i.e. tmp/job-{id}/mapper-output/ ).

For example, the Master should specify the input file is data/input/file_001.txt and the output file tmp/job-0/mapper-output/file_001.txt

Hint: See the command line package sh listed in the Libraries section. See sh.Command(…) , and the _in and _out arguments in order to funnel the input and output easily.

The Worker should be agnostic to map or reduce jobs. Regardless of the type of operation, the Worker is responsible for running the specified executable over the input files one by one, and piping to the output directory for each input file. Once a Worker has finished its job, it should send a TCP message to the Master’s main socket of the form:


At this point, you should be able to pass test_worker_3 , test_worker_4 , test_worker_5 .

Grouping – [Master + Workers]

Once all of the mappers have finished, the Master will start the “grouping” phase. This should begin right after the LAST Worker finishes the Map stage (i.e. you will get a finished message from the last Worker and the handle_msg handling that message will continue this grouping stage).

To start the group stage, the Master looks at all of the files created by the mappers, and assigns Workers to sort and merge the files. Sorting in the group stage should happen by line not by key. If there are more files than Workers, the Master should attempt to balance the files evenly among them. If there are fewer files than Workers, it is okay if some Workers sit idle during this stage. Each Worker will be responsible for merging some number of files into one larger file. The Master will then take these files, merge them into one larger file, and then partition that file into the correct number of files for the reducers. The messages sent to the Workers should look like this:


The name of the intermediate files produced – the merged files each Worker creates and the single large file the Master creates – are up to you. However, once the Master has split up the single input file into the files used for reducing, they must be named reducex , where x is the reduce task number. If there are 4 reduce jobs specified, the master should create reduce1, reduce2, reduce3, reduce4 in the grouper output directory.

Reducing – [Workers]

To the Worker, this is the same as the map stage – it doesn’t need to know if it is running a map or reduce task. The Worker just runs the executable it is told to run – the Master is responsible for making sure it tells the Worker to run the correct map or reduce executable. The output_directory in the reduce stage will always be the reducer-output folder. Again, use the same output file name as the input file.

Once a Worker has finished its job, it should send a TCP message to the Master’s main socket of the form:


Wrapping Up – [Master]

As soon as the master has received the last “finished” message for the reduce tasks for a given job, the Master should move the output files from the reducer-output directory to the final output directory specified by the original job creation message (The value specified by the output_directory key). In the final output directory, the files should be renamed finaloutputx , where x is the final output file number. If there are 4 final output files, the master should rename them finaloutput1, finaloutput2, finaloutput3, finaloutput4 . Create the output directory if it doesn’t already exist. Check the job queue for the next available job, or go back to listening for jobs if there isn’t one currently in the job queue.

Shutdown – [Master + Worker]

The Master can also receive a special message to initiate server shutdown. The shutdown message will be of the following form and will be received on the main TCP socket:


“message_type”: “shutdown”


The Master should forward this message to all of the Workers that have registered with it. The Workers, upon receiving the shutdown message, should terminate as soon as possible. If the Worker is already in the middle of executing a task, it is okay for it to complete that task before being able to handle the shutdown message as both these happen inside a single thread.

After forwarding the message to all Workers, the Master should terminate itself. At this point, you should be able to pass test_shutdown

Fault tolerance + Heartbeats – [Master + Worker]

Workers can die at any time and may not finish jobs that you send them. Your Master must accommodate for this. If a Worker misses more than 5 pings in a row, you should assume that it has died, and assign whatever work it was responsible for to another Worker machine.

Each Worker will have a heartbeat thread to send updates to Master via UDP. The messages should look like this, and should be sent every 2 seconds:


“message_type”: “heartbeat”,

“worker_pid”: int


If a Worker dies before completing all the tasks assigned to it, then all of those tasks (completed or not) should be redistributed to live Workers. At each point of the execution (mapping, grouping, reducing) the Master should attempt to evenly distribute work among available Workers. If a Worker dies while it is executing a task, the Master will have to assign that task to another Worker. You should mark the failed Worker as dead, but do not remove it from the Master’s internal data structures. This is due to constraints on the Python dictionary data structure. It can result in an error when keys are modified while iterating over the dictionary. For more info on this, please refer to this link.

Your Master should attempt to maximize concurrency, but avoid duplication – that is, don’t send the same job to different Workers until you know that the Worker who was previously assigned that task has died.

At this point, you should be able to pass test_master_3 , test_master_4 , test_worker_2 ,test_integration_1 , test_integration_2 , and test_integration_3 .

Walk-through example

See a complete example here.


To aid in writing test cases we have included a IntegrationManager class which is similar to the manager the autograder will use to test your submissions. You can find this in the starter file tests/integration_manager.py .

In addition, we have provided a simple word count map and reduce example. You can use these executables, as well as the sample data provided, and compare your server’s output with the result obtained by running:


Note that these executables can be in any language – your server should not limit us to running map and reduce jobs written in python3! To help you test this, we have also provided you with a word count solution written in bash (see section below).

Note that the autograder will swap out your Master for our Master in order to test the Worker (and vice versa). Your code should have no other dependency besides the communication spec, and the messages sent in your system must match those listed in this spec exactly.

Run the public unit tests.


Note that the -s flag has been added to the pytest command in order to also show any messages printed to stdout (such as the logging messages), to help with debugging.

Test for busy waiting

A solution that busy-waits may pass on your development machine and fail on the autograder due to a timeout. Your laptop is probably much more powerful than the restricted autograder environment, so you might not notice the performance problem locally. See the Processes, Threads and Sockets in Python Tutorial for an explanation of busy-waiting.

To detect busy waiting, time a master without any workers. After a few seconds, kill it by pressing Control – C several times. Ignore any errors or exceptions. We can tell that this solution busy-waits because the user time is similar to the real time.


Testing Fault-Tolerance

Testing for fault tolerance is a major and tricky part of this project. This section will provide some basic guidelines on how you can verify if your system handles fault-tolerance in the desired manner.

To enact the condition of a dead Worker, it is important for a Worker to die while performing a task (or in other words for the Master to realize that a Worker has missed more than 5 consecutive heartbeat messages when a still incomplete task was assigned to that Worker, post which it should declare the Worker dead and reassign the task). We have given you slow running executables of map and reduce tasks in tests/exec/wc_map_slow.sh and tests/exec/wc_reduce_slow.sh . These scripts make use of sleep statements. You can choose a sleep time that you feel gives you enough time to kill a Worker while it is execting a task.

The idea is to start your server and send a slow job to the Master. Once the task has been assigned to a Worker, since there are sleep statements in the map/reduce scripts, you should have enough time to manually kill a Worker and then see if the Master can still make forward progress (handle the dead Worker and still produce the correct output).

For example, imagine a scenario, where there are 2 Workers, each executing one slow map task task respectively. Now, the second Worker dies amidst this execution (because you manually killed the process made possible due to the sleep times in the map code). In this scenario, how many mapping tasks should the first Worker receive? How many mapping tasks should the second Worker have received? How many sorting and reducing tasks should the first and the second Worker receive? If your code gives expected answers to these questions, then you are in good shape.

Code Style

As in previous projects, all Python code should contain no errors or warnings from pycodestyle ,pydocstyle , and pylint .

You may not use any external dependencies aside from what is provided in setup.py .

Test Case Descriptions

Many of the autograder test cases in this project are visible on the autograder, but the source code is not published. We can’t publish the source code because many unit tests combine instructor code (e.g., master) with your code, (e.g., worker). This section provides a description of each test case lacking published source code.

test_master_1 :

1. Starts student master and one instructor worker

2. Verifies master received worker registration

3. Verifies master can send worker registration acknowledgement

test_master_2 :


Submitting and grading
One team member should register your group on the autograder using the create new invitation feature.

Submit a tarball to the autograder, which is linked from https://eecs485.org. Include the –disable- copyfile flag only on macOS.


Protected unit tests are visible on the autograder before the project deadline, but the source code is not published. See test case descriptions above.

Further Reading

Google’s original MapReduce paper