Optimizing Hadoop for MapReduce

By Khaled Tannir
    What do you get with a Packt Subscription?

  • Instant access to this title and 7,500+ eBooks & Videos
  • Constantly updated with 100+ new titles each month
  • Breadth and depth in over 1,000+ technologies
About this book

MapReduce is the distribution system that the Hadoop MapReduce engine uses to distribute work around a cluster by working parallel on smaller data sets. It is useful in a wide range of applications, including distributed pattern-based searching, distributed sorting, web link-graph reversal, term-vector per host, web access log stats, inverted index construction, document clustering, machine learning, and statistical machine translation.

This book introduces you to advanced MapReduce concepts and teaches you everything from identifying the factors that affect MapReduce job performance to tuning the MapReduce configuration. Based on real-world experience, this book will help you to fully utilize your cluster’s node resources to run MapReduce jobs optimally.

This book details the Hadoop MapReduce job performance optimization process. Through a number of clear and practical steps, it will help you to fully utilize your cluster’s node resources.

Starting with how MapReduce works and the factors that affect MapReduce performance, you will be given an overview of Hadoop metrics and several performance monitoring tools. Further on, you will explore performance counters that help you identify resource bottlenecks, check cluster health, and size your Hadoop cluster. You will also learn about optimizing map and reduce tasks by using Combiners and compression.

The book ends with best practices and recommendations on how to use your Hadoop cluster optimally.

Publication date:
February 2014


Chapter 1. Understanding Hadoop MapReduce

MapReduce, the popular data-intensive distributed computing model is emerging as an important programming model for large-scale data-parallel applications such as web indexing, data mining, and scientific simulation.

Hadoop is the most popular open source Java implementation of the Google's MapReduce programming model. It is already being used for large-scale data analysis tasks by many companies and is often used for jobs where low response time is critical.

Before going deep into MapReduce programming and Hadoop performance tuning, we will review the MapReduce model basics and learn about factors that affect Hadoop's performance.

In this chapter, we will cover the following:

  • The MapReduce model

  • An overview of Hadoop MapReduce

  • How MapReduce works internally

  • Factors that affect MapReduce performance


The MapReduce model

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:

  • map(): The user-defined map function is applied to all input records one by one, and for each record it outputs a list of zero or more intermediate key/value pairs, that is, <k2, v2> records. Then all <k2, v2> records are collected and reorganized so that records with the same keys (k2) are put together into a <k2, list(v2)> record.

  • reduce(): The user-defined reduce function is called once for each distinct key in the map output, <k2, list(v2)> records, and for each record the reduce function outputs zero or more <k2, v3> pairs. All <k2, v3> pairs together coalesce into the final result.


    The signatures of the map and reduce functions are as follows:

    • map(<k1, v1>) list(<k2, v2>)

    • reduce(<k2, list(v2)>) <k2, v3>

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:

  1. The input data is split into records.

  2. Map functions process these records and produce key/value pairs for each word.

  3. All key/value pairs that are output by the map function are merged together, grouped by a key, and sorted.

  4. The intermediate results are transmitted to the reduce function, which will produce the final output.

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.


An overview of Hadoop MapReduce

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.


To learn more about the Hadoop MapReduce implementation, you can browse Hadoop's official website at http://hadoop.apache.org/.

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.


Hadoop MapReduce internals

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.


Factors affecting the performance of MapReduce

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:

  • Hardware (or resources) such as CPU clock, disk I/O, network bandwidth, and memory size.

  • The underlying storage system.

  • Data size for input data, shuffle data, and output data, which are closely correlated with the runtime of a job.

  • Job algorithms (or program) such as map, reduce, partition, combine, and compress. Some algorithms may be hard to conceptualize in MapReduce, or may be inefficient to express in terms of MapReduce.

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.


In the MapReduce programming model, grouping/partitioning is a serial task! This means the framework needs to wait for all map tasks to complete before any reduce tasks can be run.

To learn more about the merge-sort algorithm, refer to the URL http://en.wikipedia.org/wiki/Merge_sort.

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:

  • The I/O mode: This is the way to retrieve data from the storage system. There atre two modes to read data from the underlying storage system:

    • Direct I/O: This is used to read directly from the local disk cache to memory through hardware controllers; therefore, no inter-process communication costs are required.

    • Streaming I/O: This allows you to read data from another running process (typically the storage system process) through certain inter-process communication schemes such as TCP/IP and JDBC.


      To enhance performance, using direct I/O may be more efficient than streaming I/O.

  • Input data parsing: This is the conversion process from raw data into the key/value pairs when data is retrieved from the storage system. The data parsing process aims to decode raw data from its native format and transform it into data objects that can be processed by a programming language such as Java.

    Input data can be decoded to (Java or other) objects so that the content can be altered after an instance is created, typically when you use a reference to an instance of the object (these objects are called mutable objects) or to objects where the content cannot be altered after it is created (called immutable objects). In the case of a million records, the immutable decoding process is significantly slower than the mutable decoding process as it may produce a huge number of immutable objects. Therefore, this can lead to poor performance of the system.

  • Input data storage: The underlining storage system must ensure a high speed access and data availability (such as HDFS, and HBase) when data is retrieved by MapReduce to be processed. If you choose to use a storage filesystem other than those recommended to be used with MapReduce, the access to the input data may potentially affect MapReduce performance.

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.

About the Author
  • Khaled Tannir

    Khaled Tannir has been working with computers since 1980. He began programming with the legendary Sinclair Zx81 and later with Commodore home computer products (Vic 20, Commodore 64, Commodore 128D, and Amiga 500). He has a Bachelor's degree in Electronics, a Master's degree in System Information Architectures, in which he graduated with a professional thesis, and completed his education with a Master of Research degree. He is a Microsoft Certified Solution Developer (MCSD) and has more than 20 years of technical experience leading the development and implementation of software solutions and giving technical presentations. He now works as an independent IT consultant and has worked as an infrastructure engineer, senior developer, and enterprise/solution architect for many companies in France and Canada. With significant experience in Microsoft .Net, Microsoft Server Systems, and Oracle Java technologies, he has extensive skills in online/offline applications design, system conversions, and multilingual applications in both domains: Internet and Desktops. He is always researching new technologies, learning about them, and looking for new adventures in France, North America, and the Middle-east. He owns an IT and electronics laboratory with many servers, monitors, open electronic boards such as Arduino, Netduino, RaspBerry Pi, and .Net Gadgeteer, and some smartphone devices based on Windows Phone, Android, and iOS operating systems. In 2012, he contributed to the EGC 2012 (International Complex Data Mining forum at Bordeaux University, France) and presented, in a workshop session, his work on "how to optimize data distribution in a cloud computing environment". This work aims to define an approach to optimize the use of data mining algorithms such as k-means and Apriori in a cloud computing environment. He is the author of RavenDB 2.x Beginner's Guide, Packt Publishing. He aims to get a PhD in Cloud Computing and Big Data and wants to learn more and more about these technologies. He enjoys taking landscape and night time photos, travelling, playing video games, creating funny electronic gadgets with Arduino/.Net Gadgeteer, and of course, spending time with his wife and family. You can reach him at contact@khaledtannir.net.

    Browse publications by this author
Optimizing Hadoop for MapReduce
Unlock this book and the full library FREE for 7 days
Start now