Optimizing Hadoop for MapReduce: This book is the perfect introduction to sophisticated concepts in MapReduce and will ensure you have the knowledge to optimize job performance. This is not an academic treatise; it's an example-driven tutorial for the real world.

MapReduce is a programming model designed for processing unstructured data by large clusters of commodity hardware and generating large datasets. It is capable of processing many terabytes of data on thousands of computing nodes in a cluster, handling failures, duplicating tasks, and aggregating results.

The MapReduce model is simple to understand. It was designed in the early 2000s by the engineers at Google Research (http://research.google.com/archive/mapreduce.html). It consists of two functions, a map function and a reduce function that can be executed in parallel on multiple machines.

To use MapReduce, the programmer writes a user-defined map function and a user-defined reduce function that expresses their desired computation. The map function reads a key/value pair, applies the user specific code, and produces results called intermediate results. Then, these intermediate results are aggregated by the reduce user-specific code that outputs the final results.

Input to a MapReduce application is organized in the records as per the input specification that will yield key/value pairs, each of which is a <k1, v1> pair.

Therefore, the MapReduce process consists of two main phases:

The MapReduce programming model is designed to be independent of storage systems. MapReduce reads key/value pairs from the underlying storage system through a reader. The reader retrieves each record from the storage system and wraps the record into a key/value pair for further processing. Users can add support for a new storage system by implementing a corresponding reader. This storage-independent design is considered to be beneficial for heterogeneous systems since it enables MapReduce to analyze data stored in different storage systems.

To understand the MapReduce programming model, let's assume you want to count the number of occurrences of each word in a given input file. Translated into a MapReduce job, the word-count job is defined by the following steps:

The overall steps of this MapReduce application are represented in the following diagram:

While aggregating key/value pairs, a massive amount of I/O and network traffic I/O can be observed. To reduce the amount of network traffic required between the map and reduce steps, the programmer can optionally perform a map-side pre-aggregation by supplying a Combiner function. Combiner functions are similar to the reduce function, except that they are not passed all the values for a given key; instead, a Combiner function emits an output value that summarizes the input values it was passed.

Hadoop is the most popular open source Java implementation of the MapReduce programming model proposed by Google. There are many other implementations of MapReduce (such as Sphere, Starfish, Riak , and so on), which implement all the features described in the Google documentation or only a subset of these features.

Hadoop consists of distributed data storage engine and MapReduce execution engine. It has been successfully used for processing highly distributable problems across a large amount of datasets using a large number of nodes. These nodes collectively form a Hadoop cluster, which in turn consists of a single master node called JobTracker, and multiple worker (or slave) nodes; each worker node is called a TaskTracker. In this framework, a user program is called a job and is divided into two steps: map and reduce.

Like in the MapReduce programing model, the user has to only define the map and reduce functions when using the Hadoop MapReduce implementation. The Hadoop MapReduce system automatically parallelizes the execution of these functions and ensures fault tolerance.

Basically, the Hadoop MapReduce framework utilizes a distributed filesystem to read and write its data. This distributed filesystem is called Hadoop Distributed File System (HDFS), which is the open source counterpart of the Google File System (GFS). Therefore, the I/O performance of a Hadoop MapReduce job strongly depends on HDFS.

HDFS consists of a master node called NameNode, and slave nodes called DataNodes. Within the HDFS, data is divided into fixed-size blocks (chunks) and spread across all DataNodes in the cluster. Each data block is typically replicated with two replicas: one placed within the same rack and the other placed outside it. NameNode keeps track of which DataNodes hold replicas of which block.

The MapReduce programing model can be used to process many large-scale data problems using one or more steps. Also, it can be efficiently implemented to support problems that deal with large amount of data using a large number of machines. In a Big Data context, the size of data processed may be so large that the data cannot be stored on a single machine.

In a typical Hadoop MapReduce framework, data is divided into blocks and distributed across many nodes in a cluster and the MapReduce framework takes advantage of data locality by shipping computation to data rather than moving data to where it is processed. Most input data blocks for MapReduce applications are located on the local node, so they can be loaded very fast, and reading multiple blocks can be done on multiple nodes in parallel. Therefore, MapReduce can achieve very high aggregate I/O bandwidth and data processing rate.

To launch a MapReduce job, Hadoop creates an instance of the MapReduce application and submits the job to the JobTracker. Then, the job is divided into map tasks (also called mappers) and reduce tasks (also called reducers).

When Hadoop launches a MapReduce job, it splits the input dataset into even-sized data blocks and uses a heartbeat protocol to assign a task. Each data block is then scheduled to one TaskTracker node and is processed by a map task.

Each task is executed on an available slot in a worker node, which is configured with a fixed number of map slots, and another fixed number of reduce slots. If all available slots are occupied, pending tasks must wait until some slots are freed up.

The TaskTracker node periodically sends its state to the JobTracker. When the TaskTracker node is idle, the JobTracker node assigns new tasks to it. The JobTracker node takes data locality into account when it disseminates data blocks. It always tries to assign a local data block to a TaskTracker node. If the attempt fails, the JobTracker node will assign a rack-local or random data block to the TaskTracker node instead.

When all map functions complete execution, the runtime system groups all intermediate pairs and launches a set of reduce tasks to produce the final results. It moves execution from the shuffle phase into the reduce phase. In this final reduce phase, the reduce function is called to process the intermediate data and write the final output.

Users often use terms with different granularities to specify Hadoop map and reduce tasks, subtasks, phases, and subphases. While the map task consists of two subtasks: map and merge, the reduce task consists of just one task. However, the shuffle and sort happen first and are done by the system. Each subtask in turn gets divided into many subphases such as read-map, spill, merge, copy-map, and reduce-write.

The processing time of input data with MapReduce may be affected by many factors. One of these factors is the algorithm you use while implementing your map and reduce functions. Other external factors may also affect the MapReduce performance. Based on our experience and observation, the following are the major factors that may affect MapReduce performance:

While running a map task, intermediate output of the shuffle subtasks is stored in a memory buffer to reduce disk I/O. However, since the size of this output may exceed that of the memory buffer and such an overflow may occur, the spill subphase is needed to flush the data into a local filesystem. This subphase may affect the MapReduce performance and is often implemented using multithreading to maximize the utility of disk I/O and to reduce the runtime of jobs.

The MapReduce programming model enables users to specify data transformation logic using their own map and reduce functions. The model does not specify how intermediate pairs produced by map functions are grouped for reduce functions to process. Therefore, the merge-sort algorithm is employed as the default grouping algorithm. However, the merge-sort algorithm is not always the most efficient algorithm, especially for analytical tasks, such as aggregation and equal-join, which do not care about the order of intermediate keys.

The MapReduce performance is based on the runtime of both map and reduce. This is because parameters such as the number of nodes in a cluster or the number of slots in a node are unmodifiable in a typical environment.

Other factors that may potentially affect the performance of MapReduce are:

When using the Hadoop framework, many factors may affect the overall system performance and the runtime of a job. These factors may be part of the Hadoop MapReduce engine or may be external to it.

The Hadoop configuration parameters usually indicate how many tasks can run concurrently and determine the runtime of a job since other factors are not modifiable after the Hadoop cluster is set up and the job starts the execution. A misconfigured Hadoop framework may underutilize the cluster resources and therefore impact the MapReduce job performance. This is due to the large number of configuration parameters that control the Hadoop framework's behavior.

A Hadoop job is often composed of many submodules that implement different algorithms, and some of these sub-modules are connected in serial, while others are connected in parallel. A misconfiguration of the Hadoop framework may impact how all internal tasks coordinate together to achieve tasks. The impact of the settings of all these parameters (which will be covered in Chapter 2, An Overview of the Hadoop Parameters) depends on the map and reduce functions' code, the cluster resources, and of course, the input data.

A MapReduce job performance can also be affected by the number of nodes in the Hadoop cluster and the available resources of all the nodes to run map and reduce tasks. Each node capacity determines the number of mapper and reducer tasks that a node can execute. Therefore, if the resources of nodes are underutilized or overutilized, it will directly impact the MapReduce tasks' performance.

In this chapter, we learned the MapReduce programing model and reviewed how this works internally. Then, we focused on Hadoop MapReduce and learned about its main components. We also covered internal and external factors that may affect Hadoop MapReduce performance.

In the next chapter, we will investigate Hadoop's tunable parameters and learn about Hadoop metrics and performance tools.