<commit_before>\documentclass{article} % For LaTeX2e
\usepackage{nips13submit_e,times}
%\usepackage{hyperref}
\usepackage[hyphens]{url}
\usepackage{alltt}
\usepackage{multirow, caption}
\usepackage{algorithm}
\usepackage{amsmath}
\usepackage{graphicx}
\usepackage[noend]{algpseudocode}
\usepackage{etoolbox}
\usepackage{hyperref}
\usepackage{fancyvrb}
\usepackage{tgcursor}
\usepackage{amsfonts}
\usepackage[ampersand]{easylist}
\newcommand\userinput[1]{\textbf{#1}}
\newcommand\comment[1]{\textit{#1}}
\newcommand\stdout[1]{\textsl{#1}}

\makeatletter
\preto{\@verbatim}{\topsep=0pt \partopsep=0pt }
\makeatother

\title{MapReduce Framework\\Final Report}

\author{
<NAME>\\
Carnegie Mellon University\\
Pittsburgh, PA 15213 \\
\texttt{<EMAIL>} \\
\And
<NAME>\\
Carnegie Mellon University\\
Pittsburgh, PA 15213 \\
\texttt{<EMAIL>} \\
}

% The \author macro works with any number of authors. There are two commands
% used to separate the names and addresses of multiple authors: \And and \AND.
%
% Using \And between authors leaves it to \LaTeX{} to determine where to break
% the lines. Using \AND forces a linebreak at that point. So, if \LaTeX{}
% puts 3 of 4 authors names on the first line, and the last on the second
% line, try using \AND instead of \And before the third author name.

\newcommand{\fix}{\marginpar{FIX}}
\newcommand{\new}{\marginpar{NEW}}

\nipsfinalcopy % Uncomment for camera-ready version

\begin{document}


\maketitle

\begin{abstract}
This report reflects our final progress on MapReduce Framework project for 15-440, Fall 2014. We will first give an overview of the framework and discuss its components one by one. Then, we will describe the lifecycle of a MapReduce job and demonstrate the communication protocol between MapReduce components. Also, we will outline how to build and test the current framework from source, and include a Developer's Guide for developing MapReduce Applications based on the framework. Eventually, we will survey some additional unimplemented features that would be desirable for a commercial package.
\end{abstract}

\section{Overview of MapReduce Framework}

\par\qquad MapReduce is a highly scalable and parallel computing model that allows application programmers to specify \emph{Map} and \emph{Reduce} methods on a distributed data set in order to complete computational tasks. According to our design, nodes on the framework have one of the three following roles: \emph{Master}, \emph{Worker}, and the \emph{Users}.
\par\qquad Essentially, \emph{Master} is the coordinating node that does not store any data or perform any computational tasks but merely work on administrative tasks like maintaining a global Job Queue and a DFS Lookup Table. On the other hand, a \emph{Worker} receives and performs tasks pushed from \emph{Master}, and also holds partitions of data on the Distributed File System. Last but not least, a \emph{User} node is a designated client of the MapReduce framework, so where an end-user is able to push data on to the DFS and execute a MapReduce job based on the data.
\par\qquad Next section contains more detailed specifications of the three components.

\section{Framework Components}

\subsection{MapReduce/DFS Master}

\par

\par Meanwhile, it is important to notice that our Master is a \emph{single point of failure} in our framework, because once it fails, all information related to Jobs, Tasks, and Distributed Files are lost, and recovery is impossible because we also lose the routing rule of partial results to the requesting client.

\subsection{MapReduce/DFS Workers}

\par\qquad They carry out all the work.

\subsection{MapReduce Users}

\par\qquad In our design, MapReduce users are specifically designated nodes that are allowed to interact with the MapReduce cluster by contacting the Master node. 

\section{Lifecycle of MapReduce Job}
Here is what happens when a MapReduce job is executed:
\subsection*{Step One: User Load Data onto DFS}
\subsection*{Step Two: User Execute Task from User CLI}
\subsection*{Step Three: Mappers Feed Results to Sorters}
\subsection*{Step Four: Sorters Feed Results to Reducers}
\subsection*{Step Five: Reducers Return Results Back to Client}

\section{Failure Handling}
\par\qquad Due to the inherent unreliability of a network, it is not unlikely that some request packages fail to go through temporarily. Also, because potentially a large number of machines can be involved in the framework, it is fairly possible that some nodes go down while running a task. Without proper failure handling, that faulty node can be a \emph{single point of failure (SPOF)}, which is what we want to avoid in a distributed system. Therefore, we adopted the following three approaches to handle failures:
\begin{easylist}[itemize]
    & Periodic Heartbeat: The master node sends periodic \texttt{ping} requests to all workers. If after a certain amount of time the worker still has not responded, we mark the worker as ``faulty'', and migrate the tasks already assigned to it to other nodes (see \emph{Task Migration} below);
    & Task Retry: When a task fails on a node for some reason, we allow the task to be re-routed to another node for re-try before marking the task as failed. With this measure, we avoid the case when a task and a worker are not compatible -- for example, when the underlying file system of that worker node has a bad sector;
    & Task Migration: After we have determined that a node is faulty, we re-distribute all tasks running on that node to other available nodes. In this case, we do not need to restart the entire Job only because of one faulty worker node.
\end{easylist}

\section{Special Considerations}
\par\qquad Following are special features that the authors think are useful for application programmers and are beyond the request of the project specification:

\subsection{Command Line Utility at RMI Servers and Users:}
    \par\qquad Different from Hadoop's approach, our MapReduce framework comes with an interactive shell for both \emph{user} and \emph{master} nodes. The shells support most functions as currently supported by Hadoop, and the advantage is that once the shell is up, running any command is only up to the communication time within the framework. In Hadoop's approach, however, a JVM needs to be started for each MapReduce/DFS operation, e.g. \texttt{hdfs dfs -ls foobar}. We are not certain why Hadoop did not choose this approach, but our alternative approach is more light-weight and gives lower latency.
\subsection{Elegant Error Catching:}
    \par\qquad Even the best application developers in the world cannot guarantee that their first attempt at writing a complex MapReduce job will work. Therefore, when an application developer feeds us buggy code that causes exceptions in mapper, sorter, or reducer, it is important that the worker node and the Master do not abort. For our framework, when a task causes an exception, the corresponding worker thread would catch it, and send \texttt{taskFailed} to the Master. Although this handling approach can still be improved (see \emph{Section 8.3}), it does guarantee that the availability of our framework is not affected by bugs in user code.
\subsection{MapReduce Elegant Shutdown:}
    \par\qquad When the Master node in the MapReduce framework shuts down, its Workers also automatically shut down after cleaning up the working directory for DFS. Meanwhile, the \emph{User} (client) nodes are kept on, but with a message \texttt{DISCONNECTED} attached to the prompt. Also, if a \emph{User} node fails to contact Master after a period of time (defined by user), its state automatically becomes \texttt{DISCONNECTED}, and the user is also able to \emph{re-connect} with the Master. This allows the user nodes to not lose their states due to temporary connection problem to the Master node.
\subsection{Worker Failure Checking with Heartbeats:}
    \par\qquad In our implementation, there is a background heartbeat thread at Master node that keeps polling all workers nodes for their statuses. Should any node fail to respond, we give the Master a certain number of chances to retry the heartbeat (defined in user configuration file). If the worker eventually fails to respond, Master re-schedules all tasks related to the node without failing the entire MapReduce job. This makes our MapReduce framework extra resilient to worker failures.
\subsection{Designated User Nodes:}
    \par\qquad In our implementation, the user nodes are specifically listed in the configuration file, i.e. the system administrator has full control over which nodes are allowed to communicate with Master node and interact with DFS or execute MapReduce jobs. This adds an extra layer of security for the framework.


\section{Developer's Guide -- Using MapReduce Framework}

\par\qquad Since the MapReduce Framework is to be used by application developers, a pre-compiled Javadoc of the project can be found in the \texttt{doc/} directory under the project root. Users can simply open \texttt{doc/index.html} and access the documentation of the entire code base including the API usage guides. The mechanism of MapReduce Framework has been explained in previous sections.

\par\qquad In order to fit a Java Object into the MapReduce Framework, say, \texttt{WordCount}, the application programmer should create three classes: \texttt{WordCountMapper} that extends the base \texttt{Mapper} class, and similarly \texttt{WordCountSorter} and \texttt{WordCountReducer}. Then, the developer should put the classes under \texttt{mapr/examples} directory, such that MapReduce users are able to execute job based on our definition of \emph{Mapper}, \emph{Sorter}, and \emph{Reducer}.

\par\qquad In general, application developers are strongly recommended to look at the sample applications in the Package \texttt{mapr.examples}, and follow the existing code. We have provided the sample code for WordCount and Grep, and other MapReduce applications can be built similarly.

\section{Dependencies, Building, and Testing}

\par\qquad The \texttt{dtzeng.yzong} handin directory will contain two sub-directories, i.e. \texttt{reports} and \texttt{RMIFramework}. In the former directory you can find this report file, and in the latter directory is the clean source code for the project.

\subsection{Dependencies}

\par\qquad This project does not have any external dependencies, and can be compiled with standard Java libraries. To build the project from scratch without using IDE, see the following section.

\subsection{Building Instructions}

\par\qquad To build the project from scratch, use the \texttt{build.sh} script provided in the handin directory at \texttt{/afs/andrew/course/15/440-f14/handin/lab3/dtzeng-yzong.x}.

\begin{Verbatim}[commandchars=\#\[\]]
gkesden@ghc11 dtzeng-yzong.x$ #userinput[chmod +x build.sh]
gkesden@ghc11 dtzeng-yzong.x$ #userinput[./build.sh]
\end{Verbatim}

\par\qquad This will create a jar file called \texttt{MapReduce.jar}.  You may copy this jar to any desired directory for testing below.

\subsection{Testing Instructions}

\par\qquad The following test routine is designed to survey most functionalities of the MapReduce framework. Feel free to experiment with different commands in MapReduce Master and MapReduce User command-line interfaces.  In our example, our facility has 1 master and 3 workers, with 2 users connected.  Let \texttt{GHC\_M} be the host of the master process, \texttt{GHC\_{W1/2/3}} the workers, and \texttt{GHC\_{U1/2}} the users.

To start up the facility, do the following:

\begin{easylist}[itemize]
 & Start up master
\begin{Verbatim}[fontsize=\scriptsize, xleftmargin=-.2in,commandchars=\#\[\]]
gkesden@#comment[GHC_M] dtzeng-yzong.x$ #userinput[java -cp MapReduce.jar mapr.master.MasterCoordinator mapr.properties]
\end{Verbatim}
 & On each of the workers
\begin{Verbatim}[fontsize=\scriptsize, xleftmargin=-.2in,commandchars=\#\[\]]
gkesden@#comment[GHC_W?] dtzeng-yzong.x$ #userinput[java -cp MapReduce.jar mapr.worker.WorkerCoordinator mapr.properties worker?]
\end{Verbatim}
 && Now that the facility has started up, the command line on the master should appear.  Make sure to connect at least 1 worker to the master before the specified timeout, otherwise the facility will shutdown automatically.  With the facility started up, you can now start the 2 users.
 & On each of the users
\begin{Verbatim}[fontsize=\scriptsize, xleftmargin=-.2in,commandchars=\#\[\]]
gkesden@#comment[GHC_U?] dtzeng-yzong.x$ #userinput[java -cp MapReduce.jar mapr.user.UserCoordinator mapr.properties user?]
\end{Verbatim}
\end{easylist}<commit_msg>report, outline for job lifecycle<commit_after>\documentclass{article} % For LaTeX2e
\usepackage{nips13submit_e,times}
%\usepackage{hyperref}
\usepackage[hyphens]{url}
\usepackage{alltt}
\usepackage{multirow, caption}
\usepackage{algorithm}
\usepackage{amsmath}
\usepackage{graphicx}
\usepackage[noend]{algpseudocode}
\usepackage{etoolbox}
\usepackage{hyperref}
\usepackage{fancyvrb}
\usepackage{tgcursor}
\usepackage{amsfonts}
\usepackage[ampersand]{easylist}
\newcommand\userinput[1]{\textbf{#1}}
\newcommand\comment[1]{\textit{#1}}
\newcommand\stdout[1]{\textsl{#1}}

\makeatletter
\preto{\@verbatim}{\topsep=0pt \partopsep=0pt }
\makeatother

\title{MapReduce Framework\\Final Report}

\author{
<NAME>\\
Carnegie Mellon University\\
Pittsburgh, PA 15213 \\
\texttt{<EMAIL>} \\
\And
<NAME>\\
Carnegie Mellon University\\
Pittsburgh, PA 15213 \\
\texttt{<EMAIL>} \\
}

% The \author macro works with any number of authors. There are two commands
% used to separate the names and addresses of multiple authors: \And and \AND.
%
% Using \And between authors leaves it to \LaTeX{} to determine where to break
% the lines. Using \AND forces a linebreak at that point. So, if \LaTeX{}
% puts 3 of 4 authors names on the first line, and the last on the second
% line, try using \AND instead of \And before the third author name.

\newcommand{\fix}{\marginpar{FIX}}
\newcommand{\new}{\marginpar{NEW}}

\nipsfinalcopy % Uncomment for camera-ready version

\begin{document}


\maketitle

\begin{abstract}
This report reflects our final progress on MapReduce Framework project for 15-440, Fall 2014. We will first give an overview of the framework and discuss its components one by one. Then, we will describe the lifecycle of a MapReduce job and demonstrate the communication protocol between MapReduce components. Also, we will outline how to build and test the current framework from source, and include a Developer's Guide for developing MapReduce Applications based on the framework. Eventually, we will survey some additional unimplemented features that would be desirable for a commercial package.
\end{abstract}

\section{Overview of MapReduce Framework}

\par\qquad MapReduce is a highly scalable and parallel computing model that allows application programmers to specify \emph{Map} and \emph{Reduce} methods on a distributed data set in order to complete computational tasks. According to our design, nodes on the framework have one of the three following roles: \emph{Master}, \emph{Worker}, and the \emph{Users}.
\par\qquad Essentially, \emph{Master} is the coordinating node that does not store any data or perform any computational tasks but merely work on administrative tasks like maintaining a global Job Queue and a DFS Lookup Table. On the other hand, a \emph{Worker} receives and performs tasks pushed from \emph{Master}, and also holds partitions of data on the Distributed File System. Last but not least, a \emph{User} node is a designated client of the MapReduce framework, so where an end-user is able to push data on to the DFS and execute a MapReduce job based on the data.
\par\qquad Next section contains more detailed specifications of the three components.

\section{Framework Components}

\subsection{MapReduce/DFS Master}

\par

\par Meanwhile, it is important to notice that our Master is a \emph{single point of failure} in our framework, because once it fails, all information related to Jobs, Tasks, and Distributed Files are lost, and recovery is impossible because we also lose the routing rule of partial results to the requesting client.

\subsection{MapReduce/DFS Workers}

\par\qquad They carry out all the work.

\subsection{MapReduce Users}

\par\qquad In our design, MapReduce users are specifically designated nodes that are allowed to interact with the MapReduce cluster by contacting the Master node. 

\section{Lifecycle of MapReduce Job}
Here is what happens when a MapReduce job is executed:
\subsection*{Step One: User Loads File onto DFS}
"upload" command on user CLI.  sends the byte contents to the master.  master takes the contents and partitions into blocks based on size specified in config.  for each block, find a random subset of workers (equal to replication factor in config), and send to each of the workers.  master updates its file trackers if successful and returns to user whether success or not.
\subsection*{Step Two: User Starts Job from User CLI}
"count" or "grep" command on user CLI.  job type and parameters are sent to the master.  master takes the parameters and splits the job into map, sort, reduce tasks and adds them to the scheduler (metadata only).  scheduler will keep track of all the dependencies and send the appropriate task metadata to the workers when ready
\subsection*{Step Three: Coordination Between Master and Workers to Complete Job}
(i feel we should combine the old step 3/4/5 because they all sort of work together, and are all concurrent at the same time)
master and workers must coordinate to make sure task dependencies are reinforced while keeping track of what else must be executed for a job.  when a map is done on a worker, it notifies the master, so the master knows that its corresponding sort is ready to be execute anytime on the intermediate file.  when a sort is done a worker, it also notifies the master, but the reduce is only ready to be launched, when all sorts on that specific worker is done.  when a reduce is done, the final result output file is done, so it also sends back the result file to the master. the master in turn will send the file back to the user, and it will be put in the user's dfs root.\\\\
(we should explain in more detail how the map, sorts, and reduce work, but this is the main outline)
\subsection*{Step Three: Mappers Feed Results to Sorters}
\subsection*{Step Four: Sorters Feed Results to Reducers}
\subsection*{Step Five: Reducers Return Results Back to Client}

\section{Failure Handling}
\par\qquad Due to the inherent unreliability of a network, it is not unlikely that some request packages fail to go through temporarily. Also, because potentially a large number of machines can be involved in the framework, it is fairly possible that some nodes go down while running a task. Without proper failure handling, that faulty node can be a \emph{single point of failure (SPOF)}, which is what we want to avoid in a distributed system. Therefore, we adopted the following three approaches to handle failures:
\begin{easylist}[itemize]
    & Periodic Heartbeat: The master node sends periodic \texttt{ping} requests to all workers. If after a certain amount of time the worker still has not responded, we mark the worker as ``faulty'', and migrate the tasks already assigned to it to other nodes (see \emph{Task Migration} below);
    & Task Retry: When a task fails on a node for some reason, we allow the task to be re-routed to another node for re-try before marking the task as failed. With this measure, we avoid the case when a task and a worker are not compatible -- for example, when the underlying file system of that worker node has a bad sector;
    & Task Migration: After we have determined that a node is faulty, we re-distribute all tasks running on that node to other available nodes. In this case, we do not need to restart the entire Job only because of one faulty worker node.
\end{easylist}

\section{Special Considerations}
\par\qquad Following are special features that the authors think are useful for application programmers and are beyond the request of the project specification:

\subsection{Command Line Utility at RMI Servers and Users:}
    \par\qquad Different from Hadoop's approach, our MapReduce framework comes with an interactive shell for both \emph{user} and \emph{master} nodes. The shells support most functions as currently supported by Hadoop, and the advantage is that once the shell is up, running any command is only up to the communication time within the framework. In Hadoop's approach, however, a JVM needs to be started for each MapReduce/DFS operation, e.g. \texttt{hdfs dfs -ls foobar}. We are not certain why Hadoop did not choose this approach, but our alternative approach is more light-weight and gives lower latency.
\subsection{Elegant Error Catching:}
    \par\qquad Even the best application developers in the world cannot guarantee that their first attempt at writing a complex MapReduce job will work. Therefore, when an application developer feeds us buggy code that causes exceptions in mapper, sorter, or reducer, it is important that the worker node and the Master do not abort. For our framework, when a task causes an exception, the corresponding worker thread would catch it, and send \texttt{taskFailed} to the Master. Although this handling approach can still be improved (see \emph{Section 8.3}), it does guarantee that the availability of our framework is not affected by bugs in user code.
\subsection{MapReduce Elegant Shutdown:}
    \par\qquad When the Master node in the MapReduce framework shuts down, its Workers also automatically shut down after cleaning up the working directory for DFS. Meanwhile, the \emph{User} (client) nodes are kept on, but with a message \texttt{DISCONNECTED} attached to the prompt. Also, if a \emph{User} node fails to contact Master after a period of time (defined by user), its state automatically becomes \texttt{DISCONNECTED}, and the user is also able to \emph{re-connect} with the Master. This allows the user nodes to not lose their states due to temporary connection problem to the Master node.
\subsection{Worker Failure Checking with Heartbeats:}
    \par\qquad In our implementation, there is a background heartbeat thread at Master node that keeps polling all workers nodes for their statuses. Should any node fail to respond, we give the Master a certain number of chances to retry the heartbeat (defined in user configuration file). If the worker eventually fails to respond, Master re-schedules all tasks related to the node without failing the entire MapReduce job. This makes our MapReduce framework extra resilient to worker failures.
\subsection{Designated User Nodes:}
    \par\qquad In our implementation, the user nodes are specifically listed in the configuration file, i.e. the system administrator has full control over which nodes are allowed to communicate with Master node and interact with DFS or execute MapReduce jobs. This adds an extra layer of security for the framework.


\section{Developer's Guide -- Using MapReduce Framework}

\par\qquad Since the MapReduce Framework is to be used by application developers, a pre-compiled Javadoc of the project can be found in the \texttt{doc/} directory under the project root. Users can simply open \texttt{doc/index.html} and access the documentation of the entire code base including the API usage guides. The mechanism of MapReduce Framework has been explained in previous sections.

\par\qquad In order to fit a Java Object into the MapReduce Framework, say, \texttt{WordCount}, the application programmer should create three classes: \texttt{WordCountMapper} that extends the base \texttt{Mapper} class, and similarly \texttt{WordCountSorter} and \texttt{WordCountReducer}. Then, the developer should put the classes under \texttt{mapr/examples} directory, such that MapReduce users are able to execute job based on our definition of \emph{Mapper}, \emph{Sorter}, and \emph{Reducer}.

\par\qquad In general, application developers are strongly recommended to look at the sample applications in the Package \texttt{mapr.examples}, and follow the existing code. We have provided the sample code for WordCount and Grep, and other MapReduce applications can be built similarly.

\section{Dependencies, Building, and Testing}

\par\qquad The \texttt{dtzeng.yzong} handin directory will contain two sub-directories, i.e. \texttt{reports} and \texttt{RMIFramework}. In the former directory you can find this report file, and in the latter directory is the clean source code for the project.

\subsection{Dependencies}

\par\qquad This project does not have any external dependencies, and can be compiled with standard Java libraries. To build the project from scratch without using IDE, see the following section.

\subsection{Building Instructions}

\par\qquad To build the project from scratch, use the \texttt{build.sh} script provided in the handin directory at \texttt{/afs/andrew/course/15/440-f14/handin/lab3/dtzeng-yzong.x}.

\begin{Verbatim}[commandchars=\#\[\]]
gkesden@ghc11 dtzeng-yzong.x$ #userinput[chmod +x build.sh]
gkesden@ghc11 dtzeng-yzong.x$ #userinput[./build.sh]
\end{Verbatim}

\par\qquad This will create a jar file called \texttt{MapReduce.jar}.  You may copy this jar to any desired directory for testing below.

\subsection{Testing Instructions}

\par\qquad The following test routine is designed to survey most functionalities of the MapReduce framework. Feel free to experiment with different commands in MapReduce Master and MapReduce User command-line interfaces.  In our example, our facility has 1 master and 3 workers, with 2 users connected.  First, change the hostnames in \texttt{mapr.properties} to reflect the desired Gates machines for testing.  Let \texttt{GHC\_M} be the host of the master process, \texttt{GHC\_{W1/2/3}} the workers, and \texttt{GHC\_{U1/2}} the users.

To start up the facility, do the following:

\begin{easylist}[itemize]
 & Start up master
\begin{Verbatim}[fontsize=\scriptsize, xleftmargin=-.2in,commandchars=\#\[\]]
gkesden@#comment[GHC_M] dtzeng-yzong.x$ #userinput[java -cp MapReduce.jar mapr.master.MasterCoordinator mapr.properties]
\end{Verbatim}
 & On each of the workers
\begin{Verbatim}[fontsize=\scriptsize, xleftmargin=-.2in,commandchars=\#\[\]]
gkesden@#comment[GHC_W?] dtzeng-yzong.x$ #userinput[java -cp MapReduce.jar mapr.worker.WorkerCoordinator mapr.properties worker?]
\end{Verbatim}
 && Now that the facility has started up, the command line on the master should appear.  Make sure to connect at least 1 worker to the master before the specified timeout, otherwise the facility will shutdown automatically.  With the facility started up, you can now start the 2 users.
 & On each of the users
\begin{Verbatim}[fontsize=\scriptsize, xleftmargin=-.2in,commandchars=\#\[\]]
gkesden@#comment[GHC_U?] dtzeng-yzong.x$ #userinput[java -cp MapReduce.jar mapr.user.UserCoordinator mapr.properties user?]
\end{Verbatim}
\end{easylist}
{'fraction_non_alphanumeric': 0.06645476049674749, 'fraction_numerical': 0.0044722057953873445, 'mean_word_length': 4.796272493573265, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2953383', 'n_tokens_mistral': 7178, 'n_tokens_neox': 6892, 'n_words': 3698}