Basic Building Blocks of Hadoop - HDFS & MapReduce

What is Hadoop

Hadoop is an open source framework from Apache and is used to store process and analyze data which are very huge in volume. Hadoop is written in Java and is not OLAP (online analytical processing). It is used for batch/offline processing. It is being used by Facebook, Yahoo, Google, Twitter, LinkedIn and many more.

Hadoop HA

The high availability feature in Hadoop ensures the availability of the Hadoop cluster without any downtime, even in unfavorable conditions like NameNode failure, DataNode failure, machine crash, etc. It means if the machine crashes, data will be accessible from another path.

History of Hadoop

Year Event

2003 Google released the paper, Google File System (GFS).

2004 Google released a white paper on Map Reduce.

2006

• Hadoop introduced.
• Hadoop 0.1.0 released.
• Yahoo deploys 300 machines and within this year reaches 600 machines.

2007

• Yahoo runs 2 clusters of 1000 machines.
• Hadoop includes HBase.

2008

• YARN JIRA opened
• Hadoop becomes the fastest system to sort 1 terabyte of data on a 900 node cluster within 209 seconds.
• Yahoo clusters loaded with 10 terabytes per day.
• Cloudera was founded as a Hadoop distributor.

2009

• Yahoo runs 17 clusters of 24,000 machines.
• Hadoop becomes capable enough to sort a petabyte.
• MapReduce and HDFS become separate subproject.

2010

• Hadoop added the support for Kerberos.
• Hadoop operates 4,000 nodes with 40 petabytes.
• Apache Hive and Pig released.

2011

• Apache Zookeeper released.
• Yahoo has 42,000 Hadoop nodes and hundreds of petabytes of storage.

2012 Apache Hadoop 1.0 version released.

2013 Apache Hadoop 2.2 version released.

2014 Apache Hadoop 2.6 version released.

2015 Apache Hadoop 2.7 version released.

2017 Apache Hadoop 3.0 version released.

2018 Apache Hadoop 3.1 version released.

Modules of Hadoop

HDFS:  Hadoop Distributed File System. Google published its paper GFS and on the basis of that HDFS was developed. It states that the files will be broken into blocks and stored in nodes over the distributed architecture.

Yarn: Yet another Resource Negotiator is used for job scheduling and manage the cluster.

Map Reduce: This is a framework which helps Java programs to do the parallel computation on data using key value pair. The Map task takes input data and converts it into a data set which can be computed in Key value pair. The output of Map task is consumed by reduce task and then the out of reducer gives the desired result.

Hadoop Architecture

The Hadoop architecture is a package of the file system, MapReduce engine and the HDFS (Hadoop Distributed File System). The MapReduce engine can be MapReduce/MR1 or YARN/MR2.

        A Hadoop cluster consists of a single master and multiple slave nodes. 

The master node includes  

• NameNode
• Secondary Namenode
• Job Tracker / Resource Manager

whereas the slave node includes

• DataNode
• Task Tracker / Node Manager

Note: 

• Namenode can communicate with Datanode vise-versa  
• Job Tracker can communicate with Task Tracker vise-versa

Hadoop Distributed File System

The Hadoop Distributed File System (HDFS) is a distributed file system for Hadoop. It contains a master/slave architecture. This architecture consist of a single NameNode performs the role of master, and multiple DataNodes performs the role of a slave.

Both NameNode and DataNode are capable enough to run on commodity machines. The Java language is used to develop HDFS. So any machine that supports Java language can easily run the NameNode and DataNode software.

NameNode

It is a single master server exist in the HDFS cluster.

As it is a single node, it may become the reason of single point failure.

It manages the file system namespace by executing an operation like the opening, renaming and closing the files.

It simplifies the architecture of the system.

DataNode

The HDFS cluster contains multiple DataNodes.

Each DataNode contains multiple data blocks.

These data blocks are used to store data.

It is the responsibility of DataNode to read and write requests from the file system's clients.

It performs block creation, deletion, and replication upon instruction from the NameNode.

Secondary Namenode

MapReduce Layer

The MapReduce comes into existence when the client application submits the MapReduce job to Job Tracker. In response, the Job Tracker sends the request to the appropriate Task Trackers. Sometimes, the TaskTracker fails or time out. In such a case, that part of the job is rescheduled.

Job Tracker 

The role of Job Tracker is to accept the MapReduce jobs from client and process the data by using NameNode.

In response, NameNode provides metadata to Job Tracker.

Task Tracker

It works as a slave node for Job Tracker.

It receives task and code from Job Tracker and applies that code on the file. This process can also be called as a Mapper.

Different layers used in Hadoop Architecture : 

The four distinct layers are 

• Application Programming interface
• Processing Framework Layer
• Cluster Resource management 
• Distributed Storage Layer

Advantages of Hadoop

Fast: In HDFS the data distributed over the cluster and are mapped which helps in faster retrieval .It is able to process terabytes of data in minutes and Peta bytes in hours.

Scalable: Hadoop cluster can be extended by just adding nodes in the cluster.

Cost Effective: Hadoop is open source and uses commodity hardware to store data so it really cost effective as compared to traditional relational database management system.

Resilient to failure: HDFS has the property with which it can replicate data over the network, so if one node is down or some other network failure happens, then Hadoop takes the other copy of data and use it. Normally, data are replicated thrice but the replication factor is configurable.

How to install Hadoop on ubuntu Step by Step

The Hadoop framework is written in Java, and its services require a compatible Java Runtime Environment (JRE) and Java Development Kit (JDK). Use the following command to update your system before initiating a new installation:

rk@ubuntu:~$ sudo apt update

At the moment, Apache Hadoop 3.x fully supports Java 8 or more. The OpenJDK  package in Ubuntu contains both the runtime environment and development kit.

rk@ubuntu:~$ sudo apt-get install default-jdk

The OpenJDK or Oracle Java version can affect how elements of a Hadoop ecosystem interact. To install a specific Java version, check out the installation process is complete

rk@ubuntu:~$ java -version; javac -version

Install the OpenSSH server and client using the following command:

rk@ubuntu:~$ sudo apt install openssh-server openssh-client -y

Create Hadoop User

Utilize the adduser command to create a new Hadoop user:

rk@ubuntu:~$ sudo adduser hduser

The username, in this example, is hduser. You are free the use any username and password you see fit. Switch to the newly created user and enter the corresponding password:

rk@ubuntu:~$ su - hduser

 

How to adduser and delete the added user from the list 

For example add the user name as rt and delete it 

Enable Pass-wordless SSH for Hadoop User

Generate an ssh keypair using keygen as shown :

hduser@ubuntu:$ ssh-keygen -t rsa -P '' -f ~/.ssh/id_rsa

Use the cat command to store the public key as authorized_keys in the ssh directory:

hduser@ubuntu:$ cat ~/.ssh/id_rsa.pub >> ~/.ssh/authorized_keys

Set the permissions for your user with the chmod command:

hduser@ubuntu:$ chmod 0600 ~/.ssh/authorized_keys 

The new user is now able to SSH without needing to enter a password every time. Verify everything is set up correctly by using the hduser user to SSH to localhost:

hduser@ubuntu:$ ssh localhost

Download and Install Hadoop on Ubuntu

Visit the official Apache Hadoop project page, and select the version of Hadoop you want to implement.

Use the provided mirror link and download the Hadoop package with the wget command:

wget https://downloads.apache.org/hadoop/common/hadoop-3.3.0/hadoop-3.3.0.tar.gz

 

Once the download is complete, extract the files to initiate the Hadoop installation:

tar xzf hadoop-3.3.0.tar.gz

Single Node Hadoop Deployment 

(Pseudo-Distributed Mode)

To explore basic commands or test applications, you can configure Hadoop on a single node pseudo-distributed mode, allows each Hadoop daemon to run as a single Java process. A Hadoop environment is configured by editing a set of configuration files:

• bashrc

• hadoop-env.sh

• core-site.xml

• hdfs-site.xml

• mapred-site-xml

• yarn-site.xml

Configure Hadoop Environment Variables (bashrc)

Edit the .bashrc shell configuration file using a text editor of your choice (we will be using nano):

hduser@ubuntu:~/hadoop-3.3.0$ sudo nano ~/.bashrc  

Define the Hadoop environment variables by adding the following content to the end of the file:

#Hadoop Related Options
export HADOOP_HOME=/home/hduser/hadoop-3.3.0
export HADOOP_INSTALL=$HADOOP_HOME
export HADOOP_MAPRED_HOME=$HADOOP_HOME
export HADOOP_COMMON_HOME=$HADOOP_HOME
export HADOOP_HDFS_HOME=$HADOOP_HOME
export YARN_HOME=$HADOOP_HOME
export HADOOP_COMMON_LIB_NATIVE_DIR=$HADOOP_HOME/lib/native
export PATH=$PATH:$HADOOP_HOME/sbin:$HADOOP_HOME/bin
export HADOOP_OPTS="-Djava.library.path=$HADOOP_HOME/lib/native"

Once you add the variables, save and exit the .bashrc file.

It is vital to apply the changes to the current running environment by using the following command:

hduser@ubuntu:~/hadoop-3.3.0$ source ~/.bashrc  

Edit hadoop-env.sh File

The hadoop-env.sh file serves as a master file to configure YARN, HDFS, MapReduce, and Hadoop-related project settings.

When setting up a single node Hadoop cluster, you need to define which Java implementation is to be utilized. Use the previously created $HADOOP_HOME variable to access the hadoop-env.sh file:

hduser@ubuntu:~/hadoop-3.3.0$ sudo nano $HADOOP_HOME/etc/hadoop/hadoop-env.sh

Uncomment the $JAVA_HOME variable (i.e., remove the # sign) and add the full path to the OpenJDK installation on your system. If you have installed the same version as presented in the first part of this tutorial, add the following line:

export JAVA_HOME=/usr/lib/jvm/java-8-openjdk-amd64

If you need help to locate the correct Java path, run the following command in your terminal window:

which javac

The resulting output provides the path to the Java binary directory.

Use the provided path to find the OpenJDK directory with the following command:

readlink -f /usr/bin/javac

The section of the path just before the /bin/javac directory needs to be assigned to the $JAVA_HOME variable.

how to install sudobers

Edit core-site.xml File

The core-site.xml file defines HDFS and Hadoop core properties.

To set up Hadoop in a pseudo-distributed mode, you need to specify the URL for your NameNode, and the temporary directory Hadoop uses for the map and reduce process.

Open the core-site.xml file in a text editor:

hduser@ubuntu:~/hadoop-3.3.0$ sudo nano /home/hduser/hadoop-3.3.0/etc/hadoop/core-site.xml  

Add the following configuration to override the default values for the temporary directory and add your HDFS URL to replace the default local file system setting:

<configuration>
<property>
  <name>hadoop.tmp.dir</name>
  <value>/home/hduser/tmpdata</value>
</property>
<property>
  <name>fs.default.name</name>
  <value>hdfs://127.0.0.1:9000</value>
</property>
</configuration>

Edit hdfs-site.xml File

The properties in the hdfs-site.xml file govern the location for storing node metadata, fsimage file, and edit log file. Configure the file by defining the NameNode and DataNode storage directories.

Additionally, the default dfs.replication value of 3 needs to be changed to 1 to match the single node setup.

Use the following command to open the hdfs-site.xml file for editing:

hduser@ubuntu:~/hadoop-3.3.0$ sudo nano /home/hduser/hadoop-3.3.0/etc/hadoop/hdfs-site.xml 

Add the following configuration to the file and, if needed, adjust the NameNode and DataNode directories to your custom locations:

<configuration>
<property>
  <name>dfs.data.dir</name>
  <value>/home/hduser/dfsdata/namenode</value>
</property>
<property>
  <name>dfs.data.dir</name>
  <value>/home/hduser/dfsdata/datanode</value>
</property>
<property>
  <name>dfs.replication</name>
  <value>1</value>
</property>
</configuration> 

Edit mapred-site.xml File

Use the following command to access the mapred-site.xml file and define MapReduce values:

hduser@ubuntu:~$ sudo nano /home/hduser/hadoop-3.3.0/etc/hadoop/mapred-site.xml

Add the following configuration to change the default MapReduce framework name value to yarn:

<configuration> 
<property> 
  <name>mapreduce.framework.name</name> 
  <value>yarn</value> 
</property> 
</configuration> 

Edit yarn-site.xml File

The yarn-site.xml file is used to define settings relevant to YARN. It contains configurations for the Node Manager, Resource Manager, Containers, and Application Master.

Open the yarn-site.xml file in a text editor:

hduser@ubuntu:~$ sudo nano /home/hduser/hadoop-3.3.0/etc/hadoop/yarn-site.xml

Append the following configuration to the file:

<configuration>
<property>
  <name>yarn.nodemanager.aux-services</name>
  <value>mapreduce_shuffle</value>
</property>
<property>
  <name>yarn.nodemanager.aux-services.mapreduce.shuffle.class</name>
  <value>org.apache.hadoop.mapred.ShuffleHandler</value>
</property>
<property>
  <name>yarn.resourcemanager.hostname</name>
  <value>127.0.0.1</value>
</property>
<property>
  <name>yarn.acl.enable</name>
  <value>0</value>
</property>
<property>
  <name>yarn.nodemanager.env-whitelist</name> 

<value>JAVA_HOME,HADOOP_COMMON_HOME,HADOOP_HDFS_HOME,HADOOP_CONF_DIR,CLASSPATH_PERPEND_DISTCACHE,HADOOP_YARN_HOME,HADOOP_MAPRED_HOME</value>

</property>
</configuration>

output :

Map Reduce

Traditional Approach :

In this approach, an enterprise will have a computer to store and process big data. Here data will be stored in an RDBMS like Oracle Database, MY SQL Server, or DB2 and sophisticated software can be written to interact with the database, process the required data, and present it to the users for analysis purposes.

Limitation :

This approach works well where we have less volume of data that can be accommodated by standard database servers, or up to the limit of the processor which is processing the data. But when it comes to dealing with huge amounts of data, it is really a tedious task to process such data through a traditional database server.

Google’s Solution :

Google solved this problem using an algorithm called Map Reduce. This algorithm divides the task into small parts and assigns those parts to many computers connected over the network, and collects the results to form the final result dataset.

The above diagram shows various commodity hardware's which could be single CPU machines or servers with higher capacity.

Map Reduce :

The Map Reduce algorithm contains two important tasks, namely Map and Reduce.

• Map (Splits, Mapping )

• Reduce (Shuffling, Reducing)

The Map task takes a set of data and converts it into another set of data, where individual elements are broken down into tuples (key-value pairs).

The Reduce task takes the output from the Map as an input and combines those data tuples (key-value pairs) into a smaller set of tuples.

The reduce task is always performed after the map job. 

The different phases in MapReduce tasks are :

Input Phase :

Here we have a Record Reader that translates each record in an input file and sends the parsed data to the mapper in the form of key-value pairs.

Map Phase :

Map is a user-defined function, which takes a series of key-value pairs and processes each one of them to generate zero or more key-value pairs.

Intermediate Keys  :

The key-value pairs generated by the mapper are known as intermediate keys.

Combiner  :

A combiner is a type of local Reducer that groups similar data from the map phase into identifiable sets. It takes the intermediate keys from the mapper as input and applies a user-defined code to aggregate the values in a small scope of one mapper. It is not a part of the main MapReduce algorithm, it is optional.

Shuffle and Sort  :

The Reducer task starts with the Shuffle and Sort step. It downloads the grouped key-value pairs onto the local machine, where the Reducer is running. The individual key-value pairs are sorted by key into a larger data list. The data list groups the equivalent keys together so that their values can be iterated easily in the Reducer task.

Reducer Phase:

The Reducer takes the grouped key-value paired data as input and runs a Reducer function on each one of them. Here, the data can be aggregated, filtered, and combined in a number of ways, and it requires a wide range of processing. Once the execution is over, it gives zero or more key-value pairs to the final step.

Output Phase :

In the output phase, we have an output formatter that translates the final key-value pairs from the Reducer function and writes them onto a file using a record writer.

MapReduce Example 1:

MapReduce Example 2:

Consider you have the following input data for your MapReduce in Big data Program

Input data:

Welcome to Hadoop 

Class Hadoop is

good Hadoop is 

bad

Final output of the MapReduce :

bad 1

Class 1

good 1

Hadoop 3

is 2

to 1

Welcome   1

The data goes through the following phases of MapReduce 

Input Splits :

An input to a MapReduce in Big Data job is divided into fixed-size pieces called input splits Input split is a chunk of the input that is consumed by a single map

Mapping :

This is the very first phase in the execution of the map-reduce program. In this phase data in each split is passed to a mapping function to produce output values. In our example, a job of the mapping phase is to count a number of occurrences of each word from input splits (more details about input-split is given below) and prepare a list in the form of <word, frequency>

Shuffling :

This phase consumes the output of the Mapping phase. Its task is to consolidate the relevant records from the Mapping phase output. In our example, the same words are clubbed together along with their respective frequency.

Reducing :

In this phase, output values from the Shuffling phase are aggregated. This phase combines values from the Shuffling phase and returns a single output value. In short, this phase summarizes the complete dataset.

Working :

The complete execution process (execution of Map and Reduce tasks, both) is controlled by two types of entities 

1. job tracker or Resource Manager: Acts like a master (responsible for complete execution of submitted job)

2. Multiple Task Trackers or Node Manager: Acts like slaves, each of them performing the job. For every job submitted for execution in the system, there is one job tracker that resides on Namenode and there are multiple task trackers which reside on Datanode.

• A job is divided into multiple tasks which are then run onto multiple data nodes in a cluster.
• It is the responsibility of the job tracker to coordinate the activity by scheduling tasks to run on different data nodes.
• Execution of individual tasks is then to look after by task tracker, which resides on every data node executing part of the job.
• Task tracker's responsibility is to send the progress report to the job tracker.
• In addition, the task tracker periodically sends a 'heartbeat' signal to the job tracker so as to notify him of the current state of the system. 
• Thus job tracker keeps track of the overall progress of each job. In the event of a task failure, the job tracker can reschedule it on a different task tracker.

Different Types of Data's used in Big Data

The data processed by Big Data solutions can be human-generated or machine-generated, although it is ultimately the responsibility of machines to generate the analytic results.

Human-generated data is the result of human interaction with systems, such as online services and digital devices.

Machine-generated data is generated by software programs and hardware devices in response to real-world events.

Examples of machine-generated data include web logs, sensor data, telemetry data, smart meter data and appliance usage data. 

Human-generated and machine-generated data can come from a variety of sources and be represented in various formats or types. 

The primary types of data are:

• structured data  ....          10%

• semi-structured data ....   10%

• unstructured data ....        80%

Structured Data :

Structured data conforms to a data model or schema and is often stored in tabular form. It is used to capture relationships between different entities and is therefore most often stored in a relational database.

Structured data is frequently generated by enterprise applications and information systems like ERP (Enterprise Resource Planning )and CRM (Customer Relationship Management). Examples of this type of data include banking transactions, invoices, and customer records.

Semi-structured Data :

Semi-structured data has a defined level of structure and consistency, but is not relational in nature. Instead, semi-structured data is hierarchical or graph-based. This kind of data is commonly stored in files that contain text.

This shows that JSON and XML files (JSON stands for JavaScript Object Notation, XML stands for eXtensible Markup Language.) are common forms of semi-structured data. Due to the textual nature of this data and its conformance to some level of structure, it is more easily processed than unstructured data.

Examples of common sources of semi-structured data include electronic data interchange (EDI) files, spreadsheets, RSS feeds and sensor data.

Unstructured Data :

Data that does not conform to a data model or data schema is known as unstructured data. It is estimated that unstructured data makes up 80% of the data within any given enterprise. Unstructured data has a faster growth rate than structured data.

This form of data is either textual or binary and often conveyed via files that are self-contained and non-relational. A text file may contain the contents of various tweets or blog postings. Binary files are often media files that contain image, audio or video data.

Unstructured data cannot be directly processed or queried using SQL. If it is required to be stored within a relational database, it is stored in a table as a Binary Large Object (BLOB). Alternatively, a Not-only SQL (NoSQL) database is a non-relational database that can be used to store unstructured data.