<figure><img src="image_1.png" /></figure>

HAL
open science

# Technology Review - Big Data, Hadoop, and MapReduce

Revant Kumar

► To cite this version:

Revant Kumar. Technology Review - Big Data, Hadoop, and MapReduce. College of Computing, Georgia Institute of Technology, USA. 2015, pp.14. [hal-04220556](https://hal.science/hal-04220556v1)

HAL Id: hal-04220556

https://hal.science/hal-04220556v1

Submitted on 28 Sep 2023

HAL is a multi-disciplinary open access
archive for the deposit and dissemination of sci-
entific research documents, whether they are pub-
lished or not. The documents may come from
teaching and research institutions in France or
abroad, or from public or private research centers.

L'archive ouverte pluridisciplinaire HAL, est
destinée au dépôt et à la diffusion de documents
scientifiques de niveau recherche, publiés ou non,
émanant des établissements d'enseignement et de
recherche français ou étrangers, des laboratoires
publics ou privés.

<figure><img src="image_2.png" /></figure>

Distributed under a Creative Commons CC BY 4.0 - Attribution - International License
Technology Review
Big Data, Hadoop, and MapReduce

Revant Kumar

MS in Computer Science
College of Computing, Georgia Institute of Technology, USA
rkumar74@gatech.edu

# 1 Big Data

Big data is a popular term used to describe the exponential growth and availability of data, both structured and unstructured.

Big data is a broad term for data sets so large or complex that traditional data processing applications are inadequate. Challenges include analysis, capture, curation, search, sharing, storage, transfer, visualization, and information privacy. The term often refers simply to the use of predictive analytics or other certain advanced methods to extract value from data, and seldom to a particular size of data set.

<figure><img src="image_3.png" /><figcaption>Figure 1: Wordle for Big Data</figcaption></figure>

Analysis of data sets can find new correlations, to “spot business trends, prevent diseases, combat crime and so on.” Scientists, practitioners of media and advertising and governments alike regularly meet difficulties with large data sets in areas including Internet search, finance and business informatics. Scientists encounter limitations in e-Science work, including meteorology, genomics, connectomics, complex physics simulations, and biological and environmental research.
Data sets grow in size in part because they are increasingly being gathered by cheap and numerous information-sensing mobile devices, aerial (remote sensing), software logs, cameras, microphones, radio-frequency identification (RFID) readers, and wireless sensor networks. The world's technological per-capita capacity to store information has roughly doubled every 40 months since the 1980s; as of 2012, every day 2.5 exabytes (2.5 × 10¹⁸) of data were created; The challenge for large enterprises is determining who should own big data initiatives that straddle the entire organization.

More data may lead to more accurate analyses. And More accurate analyses may lead to more confident decision making. And better decisions can mean greater operational efficiencies, cost reductions and reduced risk.

## 1.1 Characteristics

Big data can be described by the following characteristics:

<figure><img src="image_4.png" /><figcaption>Figure 2: Various V's of Big Data</figcaption></figure>

* **Volume** – The quantity of data that is generated is very important in this context. It is the size of the data which determines the value and potential of the data under consideration and whether it can actually be considered Big Data or not. The name Big Data itself contains a term which is related to size and hence the characteristic.

* **Variety** – The next aspect of Big Data is its variety. This means that the category to which Big Data belongs to is also a very essential fact that needs to be known by the data analysts. This helps the people, who are closely analyzing the data and are associated with it, to effectively use the data to their advantage and thus upholding the importance of the Big Data.

* **Velocity** – The term 'velocity' in the context refers to the speed of generation of data or how fast the data is generated and processed to meet the demands and the challenges which lie ahead in the path of growth and development.
* **Variability** – This is a factor which can be a problem for those who analyse the data. This refers to the inconsistency which can be shown by the data at times, thus hampering the process of being able to handle and manage the data effectively.

* **Veracity** – The quality of the data being captured can vary greatly. Accuracy of analysis depends on the veracity of the source data.

* **Complexity** – Data management can become a very complex process, especially when large volumes of data come from multiple sources. These data need to be linked, connected and correlated in order to be able to grasp the information that is supposed to be conveyed by these data. This situation, is therefore, termed as the complexity of Big Data.

## 1.2 Importance of Big Data

The hopeful vision is that organizations will be able to take data from any source, harness relevant data and analyze it to find answers that enable the following:

* Cost Reductions

* Time Reductions

* New product development and optimized offerings

* Smarter business decision making

For instance, by combining big data and high-powered analytics, it is possible to:

* Determine root causes of failures, issues and defects in near-real time, potentially saving billions of dollars annually.

* Optimize routes for many thousands of package delivery vehicles while they are on the road.

* Analyze millions of SKUs to determine prices that maximize profit and clear inventory.

* Generate retail coupons at the point of sale based on the customer's current and past purchases.

* Send tailored recommendations to mobile devices while customers are in the right area to take advantage of offers.

* Recalculate entire risk portfolios in minutes.

## 1.3 Challenges

Many organizations are concerned that the amount of amassed data is becoming so large that it is difficult to find the most valuable pieces of information.

Some of these questions are:

* What if the data volume gets so large and varied that we don't know how to deal with it?

* Do we store all the data?

* Do we analyze it all?

* How can we find out which data points are really important?

* How can we use it to our best advantage?

Until recently, organizations have been limited to using subsets of their data, or they were constrained to simplistic analyses because the sheer volumes of data overwhelmed their processing platforms. Also, on the other hand, not all business questions are better answered by bigger data.

Thus, now organizations have two choices:

* **Incorporate massive data volumes in analysis** – High-performance technologies that extract value from massive amounts of data are here today. One approach is to apply high-performance analytics to analyze the massive amounts of data using technologies such as grid computing, in-database processing and in-memory analytics.
<figure><img src="image_5.png" /><figcaption>Figure 3: Data created in 60 seconds</figcaption></figure>

* **Determine upfront which data is relevant** – Traditionally, the trend has been to store everything (some call it data hoarding) and only when we query the data do we discover what is relevant. We now have the ability to apply analytics on the front end to determine relevance based on context. This type of analysis determines which data should be included in analytical processes and what can be placed in low-cost storage for later use if needed.

## 1.4 Enabling Technologies

A number of recent technology advancements enable organizations to make the most of big data and big data analytics:

* Cheap, abundant storage.

* Faster processors.

* Affordable open source, distributed big data platforms, such as Hadoop.

* Parallel processing, clustering, MPP, virtualization, large grid environments, high connectivity and high throughputs.

* Cloud computing and other flexible resource allocation arrangements.

# 2 Hadoop

Hadoop is an open-source software framework for storing and processing big data in a distributed fashion on large clusters of commodity hardware.

Essentially, it accomplishes two tasks:

* Massive data storage

* Faster processing
<figure><img src="image_6.png" /><figcaption>Figure 4: Hadoop Logo</figcaption></figure>

Some of the characteristics of Hadoop are as follows:

* Data is divided and stored across multiple computers, and computations can be run in parallel across multiple connected machines.

* The Hadoop framework can store huge amounts of data by breaking the data into blocks and storing it on clusters of lower-cost commodity hardware.

* Hadoop processes large amounts of data in parallel across clusters of tightly connected low-cost computers for quick results.

Doug Cutting, Cloudera's Chief Architect, helped create Apache Hadoop out of necessity as data from the web exploded, and grew far beyond the ability of traditional systems to handle it. Hadoop was initially inspired by papers published by Google outlining its approach to handling an avalanche of data, and has since become the de facto standard for storing, processing and analyzing hundreds of terabytes, and even petabytes of data.

Apache Hadoop is 100% open source, and pioneered a fundamentally new way of storing and processing data. Instead of relying on expensive, proprietary hardware and different systems to store and process data, Hadoop enables distributed parallel processing of huge amounts of data across inexpensive, industry-standard servers that both store and process the data, and can scale without limits. With Hadoop, no data is too big. And in todays hyper-connected world where more and more data is being created every day, Hadoop's breakthrough advantages mean that businesses and organizations can now find value in data that was recently considered useless.

## 2.1 History behind Hadoop

As the World Wide Web grew at a dizzying pace in the late 1900s and early 2000s, search engines and indexes were created to help people find relevant information amid all of that text-based content. During the early years, search results were returned by humans. But as the number of web pages grew from dozens to millions, automation was required. Web crawlers were created, many as university-led research projects, and search engine startups took off (Yahoo, AltaVista, etc.).

One such project was Nutch, an open-source web search engine and the brainchild of Doug Cutting and Mike Cafarella. Their goal was to invent a way to return web search results faster by distributing data and calculations across different computers so multiple tasks could be accomplished simultaneously. Also during this time, another search engine project called Google was in progress. It was based on the same concept storing and processing data in a distributed, automated way so that more relevant web search results could be returned faster.

In 2006, Cutting joined Yahoo and took with him the Nutch project as well as ideas based on Google's early work with automating distributed data storage and processing. The Nutch project was divided. The web crawler portion remained as Nutch. The distributed computing and processing portion became Hadoop (named after Cutting's son's toy elephant). In 2008, Yahoo released Hadoop as an open-source project, and, today Hadoop's framework and family of technologies are
managed and maintained by the non-profit Apache Software Foundation (ASF), a global community
of software developers and contributors.

## 2.2 Importance of Hadoop

Since its inception, Hadoop has become one of the most talked about technologies. One of the top reasons (and why it was invented) is its ability to handle huge amounts of data – any kind of data – quickly. With volumes and varieties of data growing each day, especially from social media and automated sensors, that’s a key consideration for most organizations. Other reasons include:

* **Low cost** – The open-source framework is free and uses commodity hardware to store large quantities of data.

* **Computing power** – Its distributed computing model can quickly process very large volumes of data. The more computing nodes we use, the more processing power we have.

* **Scalability** – We can easily grow our system simply by adding more nodes. Little administration is required.

* **Storage flexibility** – Unlike traditional relational databases, we don't have to preprocess data before storing it. And that includes unstructured data like text, images and videos. We can store as much data as we want and decide how to use it later.

* **Inherent data protection and self-healing capabilities** – Data and application processing are protected against hardware failure. If a node goes down, jobs are automatically redirected to other nodes to make sure the distributed computing does not fail. And it automatically stores multiple copies of all data.

## 2.3 Components of Hadoop

# The Hadoop Ecosystem

<figure><img src="image_7.png" /><figcaption>Figure 5: Hadoop Ecosystem</figcaption></figure>
Hadoop components have funny names, which is sort of understandable knowing that Hadoop was the name of a yellow toy elephant owned by the son of one of its inventors. Currently three core components are included with the basic download from the Apache Software Foundation.

* **HDFS** – the Java-based distributed file system that can store all kinds of data without prior organization.

* **MapReduce** – a software programming model for processing large sets of data in parallel.

* **YARN** – a resource management framework for scheduling and handling resource requests from distributed applications.

Other components that have achieved top-level Apache project status and are available include:

* **Pig** – a platform for manipulating data stored in HDFS. It consists of a compiler for MapReduce programs and a high-level language called Pig Latin. It provides a way to perform data extractions, transformations and loading, and basic analysis without having to write MapReduce programs.

* **Hive** – a data warehousing and SQL-like query language that presents data in the form of tables. Hive programming is similar to database programming. (It was initially developed by Facebook.)

* **HBase** – a nonrelational, distributed database that runs on top of Hadoop. HBase tables can serve as input and output for MapReduce jobs.

* **Zookeeper** – an application that coordinates distributed processes.

* **Ambari** – a web interface for managing, configuring and testing Hadoop services and components.

* **Flume** – software that collects, aggregates and moves large amounts of streaming data into HDFS.

* **Sqoop** – a connection and transfer mechanism that moves data between Hadoop and relational databases.

* **Oozie** – a Hadoop job scheduler.

In addition, commercial software distributions of Hadoop are growing. Two of the most prominent (Cloudera and Hortonworks) are startups formed by the frameworks inventors. And there are plenty of others entering the Hadoop sphere. With distributions from software vendors, we pay for their version of the framework and receive additional software components, tools, training, documentation and other services.

## 2.4 Architecture

Hadoop consists of the Hadoop Common package, which provides filesystem and OS level abstractions, a MapReduce engine (either MapReduce/MR1 or YARN/MR2) and the Hadoop Distributed File System (HDFS). The Hadoop Common package contains the necessary Java Archive (JAR) files and scripts needed to start Hadoop. The package also provides source code, documentation, and a contribution section that includes projects from the Hadoop Community.

For effective scheduling of work, every Hadoop-compatible file system should provide location awareness: the name of the rack (more precisely, of the network switch) where a worker node is. Hadoop applications can use this information to run work on the node where the data is, and, failing that, on the same rack/switch, reducing backbone traffic. HDFS uses this method when replicating data to try to keep different copies of the data on different racks. The goal is to reduce the impact of a rack power outage or switch failure, so that even if these events occur, the data may still be readable.

A small Hadoop cluster includes a single master and multiple worker nodes. The master node consists of a JobTracker, TaskTracker, NameNode and DataNode. A slave or worker node acts as both a DataNode and TaskTracker, though it is possible to have data-only worker nodes and compute-only worker nodes. These are normally used only in nonstandard applications.
<figure><img src="image_8.png" /><figcaption>Figure 6: A multi-node Hadoop cluster</figcaption></figure>

Hadoop requires Java Runtime Environment (JRE) 1.6 or higher. The standard startup and shutdown scripts require that Secure Shell (ssh) be set up between nodes in the cluster.

In a larger cluster, the HDFS is managed through a dedicated NameNode server to host the file system index, and a secondary NameNode that can generate snapshots of the namenode's memory structures, thus preventing file-system corruption and reducing loss of data. Similarly, a standalone JobTracker server can manage job scheduling. In clusters where the Hadoop MapReduce engine is deployed against an alternate file system, the NameNode, secondary NameNode, and DataNode architecture of HDFS are replaced by the file-system-specific equivalents.

### 2.4.1 Hadoop Distributed File System (HDFS)

The Hadoop distributed file system (HDFS) is a distributed, scalable, and portable file-system written in Java for the Hadoop framework. A Hadoop cluster has nominally a single namenode plus a cluster of datanodes, although redundancy options are available for the namenode due to its criticality. Each datanode serves up blocks of data over the network using a block protocol specific to HDFS. The file system uses TCP/IP sockets for communication. Clients use remote procedure call (RPC) to communicate between each other.

HDFS stores large files (typically in the range of gigabytes to terabytes) across multiple machines. It achieves reliability by replicating the data across multiple hosts, and hence theoretically does not require RAID storage on hosts (but to increase I/O performance some RAID configurations are still useful). With the default replication value, 3, data is stored on three nodes: two on the same rack, and one on a different rack. Data nodes can talk to each other to rebalance data, to move copies around, and to keep the replication of data high. HDFS is not fully POSIX-compliant, because the requirements for a POSIX file-system differ from the target goals for a Hadoop application. The trade-off of not having a fully POSIX-compliant file-system is increased performance for data throughput and support for non-POSIX operations such as Append.

### 2.4.2 JobTracker and TaskTracker: the MapReduce engine

Above HDFS comes the MapReduce Engine, which consists of one JobTracker, to which client applications submit MapReduce jobs. The JobTracker pushes work out to available TaskTracker
HDFS Architecture

<figure><img src="image_9.png" /><figcaption>Figure 7: Hadoop Distributed File System</figcaption></figure>

nodes in the cluster, striving to keep the work as close to the data as possible. With a rack-aware file
system, the JobTracker knows which node contains the data, and which other machines are nearby.
If the work cannot be hosted on the actual node where the data resides, priority is given to nodes
in the same rack. This reduces network traffic on the main backbone network. If a TaskTracker
fails or times out, that part of the job is rescheduled. The TaskTracker on each node spawns off a
separate Java Virtual Machine process to prevent the TaskTracker itself from failing if the running
job crashes its JVM. A heartbeat is sent from the TaskTracker to the JobTracker every few minutes
to check its status. The Job Tracker and TaskTracker status and information is exposed by Jetty and
can be viewed from a web browser.

Known limitations of this approach are:

* The allocation of work to TaskTrackers is very simple. Every TaskTracker has a number of available slots (such as “4 slots”). Every active map or reduce task takes up one slot. The Job Tracker allocates work to the tracker nearest to the data with an available slot. There is no consideration of the current system load of the allocated machine, and hence its actual availability.

* If one TaskTracker is very slow, it can delay the entire MapReduce job especially towards the end of a job, where everything can end up waiting for the slowest task. With speculative execution enabled, however, a single task can be executed on multiple slave nodes.

2.4.3 Scheduling

By default Hadoop uses FIFO, and optionally 5 scheduling priorities to schedule jobs from a work
queue. In version 0.19 the job scheduler was refactored out of the JobTracker, while adding the
ability to use an alternate scheduler (such as the Fair scheduler or the Capacity scheduler).
<figure><img src="image_10.png" /><figcaption>Figure 8: JobTracker and TaskTracker: the MapReduce engine</figcaption></figure>

## 2.5 Getting data into Hadoop

There are numerous ways to get data into Hadoop. Here are just a few:

* We can load files to the file system using simple Java commands, and HDFS takes care of making multiple copies of data blocks and distributing those blocks over multiple nodes in Hadoop.

* If we have a large number of files, a shell script that will run multiple "put" commands in parallel will speed up the process. We don't have to write MapReduce code.

* Create a cron job to scan a directory for new files and "put" them in HDFS as they show up. This is useful for things like downloading email at regular intervals.

* Mount HDFS as a file system and simply copy or write files there.

* Use Sqoop to import structured data from a relational database to HDFS, Hive and HBase. It can also extract data from Hadoop and export it to relational databases and data warehouses.

* Use Flume to continuously load data from logs into Hadoop.

## 2.6 Popular Uses of Hadoop

Now, Hadoop is being looked as the next big data platform by many organizations, and thus they are going beyond its original goal of searching millions (or billions) of web pages and returning relevant results.

Some of the more popular uses for the framework today are as follows:

* **Low-cost storage and active data archive** – The modest cost of commodity hardware makes Hadoop useful for storing and combining big data such as transactional, social media, and E-commerce data.
## 3.1 Programming Model

<figure><img src="image_11.png" /><figcaption>Figure 9: MapReduce Workflow</figcaption></figure>

The computation takes a set of input key/value pairs, and produces a set of output key/value pairs. The user of the MapReduce library expresses the computation as two functions: Map and Reduce.

Map, written by the user, takes an input pair and produces a set of intermediate key/value pairs. The MapReduce library groups together all intermediate values associated with the same intermediate key I and passes them to the Reduce function.

The Reduce function, also written by the user, accepts an intermediate key I and a set of values for that key. It merges together these values to form a possibly smaller set of values. Typically just zero or one output value is produced per Reduce invocation. The intermediate values are supplied to the users reduce function via an iterator. This allows us to handle lists of values that are too large to fit in memory.

Conceptually the map and reduce functions supplied by the user have associated types:

$$ 
\begin{aligned}
map(k_1, v_1) & \rightarrow list(k_2, v_2) \\
reduce(k_2, list(v_2)) & \rightarrow list(v_2)
\end{aligned}
$$

Thus, the input keys and values are drawn from a different domain than the output keys and values. Furthermore, the intermediate keys and values are from the same domain as the output keys and values.

## 3.2 Execution Overview

The Map invocations are distributed across multiple machines by automatically partitioning the input data into a set of M splits. The input splits can be processed in parallel by different machines. Reduce invocations are distributed by partitioning the intermediate key space into R pieces using
<figure><img src="image_12.png" /><figcaption>Figure 10: MapReduce Execution Overview</figcaption></figure>

a partitioning function (e.g., hash(key) mod R). The number of partitions (R) and the partitioning function are specified by the user.

Figure 10 shows the overall flow of a MapReduce operation. When the user program calls the MapReduce function, the following sequence of actions occurs (the numbered labels in Figure 10 correspond to the numbers in the list below):

1. The MapReduce library in the user program first splits the input files into M pieces of typically 16 megabytes to 64 megabytes (MB) per piece (controllable by the user via an optional parameter). It then starts up many copies of the program on a cluster of machines.

2. One of the copies of the program is special – the master. The rest are workers that are assigned work by the master. There are M map tasks and R reduce tasks to assign. The master picks idle workers and assigns each one a map task or a reduce task.

3. A worker who is assigned a map task reads the contents of the corresponding input split. It parses key/value pairs out of the input data and passes each pair to the user-defined Map function. The intermediate key/value pairs produced by the Map function are buffered in memory.

4. Periodically, the buffered pairs are written to local disk, partitioned into R regions by the partitioning function. The locations of these buffered pairs on the local disk are passed back to the master, who is responsible for forwarding these locations to the reduce workers.

5. When a reduce worker is notified by the master about these locations, it uses remote procedure calls to read the buffered data from the local disks of the map workers. When a reduce worker has read all intermediate data, it sorts it by the intermediate keys so that all occurrences of the same key are grouped together. The sorting is needed because typically many different keys map to the same reduce task. If the amount of intermediate data is too large to fit in memory, an external sort is used.

6. The reduce worker iterates over the sorted intermediate data and for each unique intermediate key encountered, it passes the key and the corresponding set of intermediate values to
the users Reduce function. The output of the Reduce function is appended to a final output file for this reduce partition.

7. When all map tasks and reduce tasks have been completed, the master wakes up the user program. At this point, the MapReduce call in the user program returns back to the user code.

After successful completion, the output of the mapreduce execution is available in the R output files (one per reduce task, with file names as specified by the user). Typically, users do not need to combine these R output files into one file – they often pass these files as input to another MapReduce call, or use them from another distributed application that is able to deal with input that is partitioned into multiple files.

### 3.3 Uses

MapReduce is useful in a wide range of applications, including distributed pattern-based searching, distributed sorting, web link-graph reversal, Singular Value Decomposition, web access log stats, inverted index construction, document clustering, machine learning, and statistical machine translation. Moreover, the MapReduce model has been adapted to several computing environments like multi-core and many-core systems, desktop grids, volunteer computing environments, dynamic cloud environments, and mobile environments.

At Google, MapReduce was used to completely regenerate Google's index of the World Wide Web. It replaced the old ad hoc programs that updated the index and ran the various analyses. Development at Google has since moved on to technologies such as Percolator, Flume and MillWheel that offer streaming operation and updates instead of batch processing, to allow integrating "live" search results without rebuilding the complete index.

MapReduce's stable inputs and outputs are usually stored in a distributed file system. The transient data is usually stored on local disk and fetched remotely by the reducers.

## 4 Conclusion

With data growing so rapidly and the rise of unstructured data accounting for 90% of the data today, the time has come for enterprises to re-evaluate their approach to data storage, management and analytics. Legacy systems will remain necessary for specific high-value, low-volume workloads, and complement the use of Hadoop – optimizing the data management structure in the organization by putting the right Big Data workloads in the right systems. The cost-effectiveness, scalability, and streamlined architectures of Hadoop will make the technology more and more attractive.

## 5 References

[1] Jeffrey Dean & Sanjay Ghemawat. "MapReduce: Simplified Data Processing on Large Clusters"

[2] http://www.sas.com/en_us/insights/big-data/hadoop.html

[3] http://www.sas.com/en_us/insights/big-data/what-is-big-data.html

[4] http://www.cloudera.com/content/cloudera/en/about/hadoop-and-big-data.html

[5] Jeffrey Dean and Sanjay Ghemawat. "MapReduce: A Flexible Data Processing Tool"

[6] http://en.wikipedia.org/wiki/Big_data

[7] Hadoop (for batch indexing): http://hadoop.apache.org/

[8] HBase - The Definitive Guide: Random Access to Your Planet-Size Data

[9] Hive: A Petabyte Scale Data Warehouse Using Hadoop. In ICDE, 2010.

[10] http://en.wikipedia.org/wiki/Apache_Hadoop

[11] http://en.wikipedia.org/wiki/MapReduce