ASSIGNMENT

Introduction to Big Data

Distributed File System
Configuration – Hadoop,SPARK,Pig,Hive

What is Mapreduce?
MapReduce is a programming paradigm that runs in the background of Hadoop
to provide scalability and easy data-processing solutions. This tutorial explains
the features of MapReduce and how it works to analyze Big Data.
What is Hadoop?
Hadoop is an open-source framework that allows to store and process big data in
a distributed environment across clusters of computers using simple
programming models. It is designed to scale up from single servers to thousands
of machines, each offering local computation and storage.

What is Big Data?

Big Data is a collection of large datasets that cannot be processed using
traditional computing techniques. For example, the volume of data
Facebook or Youtube need require it to collect and manage on a daily
basis, can fall under the category of Big Data. However, Big Data is not
only about scale and volume, it also involves one or more of the following
aspects − Velocity, Variety, Volume, and Complexity.

Why MapReduce?
Traditional Enterprise Systems normally have a centralized server to store
and process data. The following illustration depicts a schematic view of a
traditional enterprise system. Traditional model is certainly not suitable to
process huge volumes of scalable data and cannot be accommodated by
standard database servers. Moreover, the centralized system creates too
much of a bottleneck while processing multiple files simultaneously.
 Google solved this bottleneck issue using an algorithm called MapReduce.
MapReduce divides a task into small parts and assigns them to many
computers. Later, the results are collected at one place and integrated to
form the result dataset.

Map Reduce Algorithm

How MapReduce Works?

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

 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.

Let us now take a close look at each of the phases and try to understand
their significance.
 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 − 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 − They 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 − 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.

Let us try to understand the two tasks Map &f Reduce with the help of a
small diagram −
 MapReduce-Example
Let us take a real-world example to comprehend the power of MapReduce.
Twitter receives around 500 million tweets per day, which is nearly 3000
tweets per second. The following illustration shows how Tweeter manages
its tweets with the help of MapReduce.

As shown in the illustration, the MapReduce algorithm performs the

following actions −

 Tokenize − Tokenizes the tweets into maps of tokens and writes them as
key-value pairs.
 Filter − Filters unwanted words from the maps of tokens and writes the
filtered maps as key-value pairs.
 Count − Generates a token counter per word.
 Aggregate Counters − Prepares an aggregate of similar counter values
into small manageable units.

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

 The map task is done by means of Mapper Class

 The reduce task is done by means of Reducer Class.

Mapper class takes the input, tokenizes it, maps and sorts it. The output of
Mapper class is used as input by Reducer class, which in turn searches
matching pairs and reduces them.

MapReduce implements various mathematical algorithms to divide a task

into small parts and assign them to multiple systems. In technical terms,
MapReduce algorithm helps in sending the Map & Reduce tasks to
appropriate servers in a cluster.

These mathematical algorithms may include the following −

 Sorting
 Searching
 Indexing
 TF-IDF

Sorting
Sorting is one of the basic MapReduce algorithms to process and analyze
data. MapReduce implements sorting algorithm to automatically sort the
output key-value pairs from the mapper by their keys.

 Sorting methods are implemented in the mapper class itself.

 In the Shuffle and Sort phase, after tokenizing the values in the mapper
class, the Context class (user-defined class) collects the matching valued
keys as a collection.
 To collect similar key-value pairs (intermediate keys), the Mapper class
takes the help of RawComparator class to sort the key-value pairs.
 The set of intermediate key-value pairs for a given Reducer is
automatically sorted by Hadoop to form key-values (K2, {V2, V2, }) before
they are presented to the Reducer.

Searching
Searching plays an important role in MapReduce algorithm. It helps in the
combiner phase (optional) and in the Reducer phase. Let us try to
understand how Searching works with the help of an example.

Example
The following example shows how MapReduce employs Searching
algorithm to find out the details of the employee who draws the highest
salary in a given employee dataset.

 Let us assume we have employee data in four different files − A, B, C, and

D. Let us also assume there are duplicate employee records in all four files
because of importing the employee data from all database tables
repeatedly. See the following illustration.

 The Map phase processes each input file and provides the employee
data in key-value pairs (<k, v> : <emp name, salary>). See the following
illustration.

 The combiner phase (searching technique) will accept the input from
the Map phase as a key-value pair with employee name and salary. Using
searching technique, the combiner will check all the employee salary to
find the highest salaried employee in each file. See the following snippet.
<k: employee name, v: salary>
Max= the salary of an first employee. Treated as max salary
 if(v(second employee).salary > Max){
Max = v(salary);
}

else{
Continue checking;
}

The expected result is as follows −

<satish, <gopal, <kiran, <manisha,

26000> 50000> 45000> 45000>

 Reducer phase − Form each file, you will find the highest salaried
employee. To avoid redundancy, check all the <k, v> pairs and eliminate
duplicate entries, if any. The same algorithm is used in between the four
<k, v> pairs, which are coming from four input files. The final output
should be as follows −
<gopal, 50000>

Indexing
Normally indexing is used to point to a particular data and its address. It
performs batch indexing on the input files for a particular Mapper.

The indexing technique that is normally used in MapReduce is known

as inverted index. Search engines like Google and Bing use inverted
indexing technique. Let us try to understand how Indexing works with the
help of a simple example.
Example
The following text is the input for inverted indexing. Here T[0], T[1], and
t[2] are the file names and their content are in double quotes.

T[0] = "it is what it is"

T[1] = "what is it"
T[2] = "it is a banana"

After applying the Indexing algorithm, we get the following output −

"a": {2}
"banana": {2}
"is": {0, 1, 2}
"it": {0, 1, 2}
"what": {0, 1}

Here "a": {2} implies the term "a" appears in the T[2] file. Similarly, "is":
{0, 1, 2} implies the term "is" appears in the files T[0], T[1], and T[2].

TF-IDF
TF-IDF is a text processing algorithm which is short for Term Frequency −
Inverse Document Frequency. It is one of the common web analysis
algorithms. Here, the term 'frequency' refers to the number of times a
term appears in a document.

Term Frequency (TF)

It measures how frequently a particular term occurs in a document. It is
calculated by the number of times a word appears in a document divided
by the total number of words in that document.

TF(the) = (Number of times term the the appears in a document) /

(Total number of terms in the document)

Inverse Document Frequency (IDF)

It measures the importance of a term. It is calculated by the number of
documents in the text database divided by the number of documents
where a specific term appears.

While computing TF, all the terms are considered equally important. That
means, TF counts the term frequency for normal words like is, a, what,
etc. Thus we need to know the frequent terms while scaling up the rare
ones, by computing the following −

IDF(the) = log_e(Total number of documents / Number of documents

with term the in it).

The algorithm is explained below with the help of a small example.

Example
Consider a document containing 1000 words, wherein the
word hive appears 50 times. The TF for hive is then (50 / 1000) = 0.05.
 Now, assume we have 10 million documents and the word hive appears
in 1000 of these. Then, the IDF is calculated as log(10,000,000 / 1,000) =
4.

The TF-IDF weight is the product of these quantities − 0.05 4 = 0.20.

Introduction to Hadoop Distributed File

System(HDFS)




With growing data velocity the data size easily outgrows the storage limit
of a machine. A solution would be to store the data across a network of
machines. Such filesystems are called distributed filesystems. Since
data is stored across a network all the complications of a network come
in.
This is where Hadoop comes in. It provides one of the most reliable
filesystems. HDFS (Hadoop Distributed File System) is a unique design
that provides storage for extremely large files with streaming data access
pattern, and it runs on commodity hardware. Let's elaborate on the
terms:
 Extremely large files: Here, we are talking about the data in a range
of petabytes (1000 TB).
 Streaming Data Access Pattern: HDFS is designed on principle
of write-once and read-many-times. Once data is written large portions
of dataset can be processed any number times.
 Commodity hardware: Hardware that is inexpensive and easily
available in the market. This is one of the features that especially
distinguishes HDFS from other file systems.
Nodes: Master-slave nodes typically form the HDFS cluster.
1. NameNode(MasterNode):

 Manages all the slave nodes and assigns work to them.

 It executes filesystem namespace operations like opening, closing,
and renaming files and directories.
 It should be deployed on reliable hardware that has a high
configuration. not on commodity hardware.
2. DataNode(SlaveNode):

 Actual worker nodes do the actual work like reading, writing,

processing, etc.
 They also perform creation, deletion, and replication upon
instruction from the master.
 They can be deployed on commodity hardware.
HDFS daemons: Daemons are the processes running in the
background.
 Namenodes:

o Run on the master node.

o Store metadata (data about data) like file path, the number of
blocks, block Ids. etc.
o Requires a high amount of RAM.
o Store meta-data in RAM for fast retrieval i.e to reduce seek
time. Though a persistent copy of it is kept on disk.
 DataNodes:

o Run on slave nodes.

o Require high memory as data is actually stored here.
Data storage in HDFS: Now let's see how the data is stored in a
distributed manner.

Lets assume that 100TB file is inserted, then masternode(namenode) will

first divide the file into blocks of 10TB (default size is 128 MB in Hadoop
2.x and above). Then these blocks are stored across different
datanodes(slavenode). Datanodes(slavenode) replicate the blocks among
themselves and the information of what blocks they contain is sent to the
master. Default replication factor is 3 means for each block 3 replicas are
created (including itself). In [Link] we can increase or decrease the
replication factor i.e we can edit its configuration here.

Note: MasterNode has the record of everything, it knows the location and
info of each and every single data nodes and the blocks they contain, i.e.
nothing is done without the permission of masternode.

Why divide the file into blocks?

Answer: Let's assume that we don't divide, now it's very difficult to store a
100 TB file on a single machine. Even if we store, then each read and
write operation on that whole file is going to take very high seek time. But
if we have multiple blocks of size 128MB then its become easy to perform
various read and write operations on it compared to doing it on a whole
file at once. So we divide the file to have faster data access i.e. reduce
seek time.

Why replicate the blocks in data nodes while storing?

Answer: Let's assume we don't replicate and only one yellow block is
present on datanode D1. Now if the data node D1 crashes we will lose the
block and which will make the overall data inconsistent and faulty. So we
replicate the blocks to achieve fault-tolerance.

Terms related to HDFS:

 HeartBeat : It is the signal that datanode continuously sends to
namenode. If namenode doesn't receive heartbeat from a datanode
then it will consider it dead.
 Balancing : If a datanode is crashed the blocks present on it will be
gone too and the blocks will be under-replicated compared to the
remaining blocks. Here master node(namenode) will give a signal to
datanodes containing replicas of those lost blocks to replicate so that
overall distribution of blocks is balanced.
 Replication:: It is done by datanode.

Note: No two replicas of the same block are present on the same
datanode.

Features:
 Distributed data storage.
 Blocks reduce seek time.
 The data is highly available as the same block is present at multiple
datanodes.
 Even if multiple datanodes are down we can still do our work, thus
making it highly reliable.
 High fault tolerance.
Limitations: Though HDFS provide many features there are some areas
where it doesn't work well.
 Low latency data access: Applications that require low-latency
access to data i.e in the range of milliseconds will not work well with
HDFS, because HDFS is designed keeping in mind that we need high-
throughput of data even at the cost of latency.
 Small file problem: Having lots of small files will result in lots of seeks
and lots of movement from one datanode to another datanode to
retrieve each small file, this whole process is a very inefficient data
access pattern.

Hadoop – Shell Commands

There are many more commands in "$HADOOP_HOME/bin/hadoop

fs" than are demonstrated here, although these basic operations will get
you started. Running ./bin/hadoop dfs with no additional arguments will
list all the commands that can be run with the FsShell system.
Furthermore, $HADOOP_HOME/bin/hadoop fs -help commandName
will display a short usage summary for the operation in question, if you
are stuck.

A table of all the operations is shown below. The following conventions are
used for parameters −

"<path>" means any file or directory name.

"<path>..." means one or more file or directory names.
"<file>" means any filename.
"<src>" and "<dest>" are path names in a directed operation.
"<localSrc>" and "<localDest>" are paths as above, but on the
local file system.

All other files and path names refer to the objects inside HDFS.
Sr.N
Command & Description
o

-ls <path>
1 Lists the contents of the directory specified by path, showing the names, permissions, owner, size and
modification date for each entry.

-lsr <path>
2
Behaves like -ls, but recursively displays entries in all subdirectories of path.

-du <path>
3 Shows disk usage, in bytes, for all the files which match path; filenames are reported with the full HDFS
protocol prefix.

-dus <path>
4
Like -du, but prints a summary of disk usage of all files/directories in the path.

-mv <src><dest>
5
Moves the file or directory indicated by src to dest, within HDFS.

-cp <src> <dest>

6
Copies the file or directory identified by src to dest, within HDFS.

-rm <path>
7
Removes the file or empty directory identified by path.

-rmr <path>
8 Removes the file or directory identified by path. Recursively deletes any child entries (i.e., files or
subdirectories of path).

-put <localSrc> <dest>

9
Copies the file or directory from the local file system identified by localSrc to dest within the DFS.

-copyFromLocal <localSrc> <dest>

10
Identical to -put

-moveFromLocal <localSrc> <dest>

11 Copies the file or directory from the local file system identified by localSrc to dest within HDFS, and
then deletes the local copy on success.

-get [-crc] <src> <localDest>

12
Copies the file or directory in HDFS identified by src to the local file system path identified by localDest.

-getmerge <src> <localDest>

13 Retrieves all files that match the path src in HDFS, and copies them to a single, merged file in the local
file system identified by localDest.

-cat <filen-ame>
14
Displays the contents of filename on stdout.

-copyToLocal <src> <localDest>

15
Identical to -get

-moveToLocal <src> <localDest>

16
Works like -get, but deletes the HDFS copy on success.

-mkdir <path>
17 Creates a directory named path in HDFS.
Creates any parent directories in path that are missing (e.g., mkdir -p in Linux).
 Anatomy of File Read and Write in HDFS




Big data is nothing but a collection of data sets that are large, complex, and
which are difficult to store and process using available data management
tools or traditional data processing applications. Hadoop is a framework
(open source) for writing, running, storing, and processing large datasets in
a parallel and distributed manner. It is a solution that is used to overcome
the challenges faced by big data.

Hadoop has two components:

 HDFS (Hadoop Distributed File System)

 YARN (Yet Another Resource Negotiator)

In this article, we focus on one of the components of Hadoop i.e., HDFS

and the anatomy of file reading and file writing in HDFS. HDFS is a file
system designed for storing very large files (files that are hundreds of
megabytes, gigabytes, or terabytes in size) with streaming data access,
running on clusters of commodity hardware(commonly available hardware
that can be obtained from various vendors). In simple terms, the storage
unit of Hadoop is called HDFS.

Some of the characteristics of HDFS are:

 Fault-Tolerance
 Scalability
 Distributed Storage
 Reliability
 High availability
 Cost-effective
 High throughput

Building Blocks of Hadoop:

1. Name Node
2. Data Node
3. Secondary Name Node (SNN)
4. Job Tracker
5. Task Tracker

Anatomy of File Read in HDFS

Let's get an idea of how data flows between the client interacting with
HDFS, the name node, and the data nodes with the help of a diagram.
Consider the figure:

Step 1: The client opens the file it wishes to read by calling open() on the
File System Object(which for HDFS is an instance of Distributed File
System).

Step 2: Distributed File System( DFS) calls the name node, using remote
procedure calls (RPCs), to determine the locations of the first few blocks in
the file. For each block, the name node returns the addresses of the data
nodes that have a copy of that block. The DFS returns an
FSDataInputStream to the client for it to read data from.
FSDataInputStream in turn wraps a DFSInputStream, which manages the
data node and name node I/O.

Step 3: The client then calls read() on the stream. DFSInputStream, which
has stored the info node addresses for the primary few blocks within the
file, then connects to the primary (closest) data node for the primary block
in the file.

Step 4: Data is streamed from the data node back to the client, which calls
read() repeatedly on the stream.

Step 5: When the end of the block is reached, DFSInputStream will close
the connection to the data node, then finds the best data node for the next
block. This happens transparently to the client, which from its point of view
is simply reading an endless stream. Blocks are read as, with the
DFSInputStream opening new connections to data nodes because the
client reads through the stream. It will also call the name node to retrieve
the data node locations for the next batch of blocks as needed.

Step 6: When the client has finished reading the file, a function is called,
close() on the FSDataInputStream.

Anatomy of File Write in HDFS

Next, we'll check out how files are written to HDFS. Consider figure 1.2 to
get a better understanding of the concept.
Note: HDFS follows the Write once Read many times model. In HDFS we
cannot edit the files which are already stored in HDFS, but we can append
data by reopening the files.
Step 1: The client creates the file by calling create() on
DistributedFileSystem(DFS).

Step 2: DFS makes an RPC call to the name node to create a new file in
the file system's namespace, with no blocks associated with it. The name
node performs various checks to make sure the file doesn't already exist
and that the client has the right permissions to create the file. If these
checks pass, the name node prepares a record of the new file; otherwise,
the file can't be created and therefore the client is thrown an error i.e.
IOException. The DFS returns an FSDataOutputStream for the client to
start out writing data to.

Step 3: Because the client writes data, the DFSOutputStream splits it into
packets, which it writes to an indoor queue called the info queue. The data
queue is consumed by the DataStreamer, which is liable for asking the
name node to allocate new blocks by picking an inventory of suitable data
nodes to store the replicas. The list of data nodes forms a pipeline, and
here we'll assume the replication level is three, so there are three nodes in
the pipeline. The DataStreamer streams the packets to the primary data
node within the pipeline, which stores each packet and forwards it to the
second data node within the pipeline.

Step 4: Similarly, the second data node stores the packet and forwards it to
the third (and last) data node in the pipeline.
 Step 5: The DFSOutputStream sustains an internal queue of packets that
are waiting to be acknowledged by data nodes, called an "ack queue".

Step 6: This action sends up all the remaining packets to the data node
pipeline and waits for acknowledgments before connecting to the name
node to signal whether the file is complete or not.

HDFS follows Write Once Read Many models. So, we can't edit files that
are already stored in HDFS, but we can include them by again reopening
the file. This design allows HDFS to scale to a large number of concurrent
clients because the data traffic is spread across all the data nodes in the
cluster. Thus, it increases the availability, scalability, and throughput of the
system.

Hadoop - Daemons and Their Features





Daemons mean Process. Hadoop Daemons are a set of processes that

run on Hadoop. Hadoop is a framework written in Java, so all these
processes are Java Processes.

Apache Hadoop 2 consists of the following Daemons:

 NameNode
 DataNode
 Secondary Name Node
 Resource Manager
 Node Manager
Namenode, Secondary NameNode, and Resource Manager work on a
Master System while the Node Manager and DataNode work on the Slave
machine.
1. NameNode
NameNode works on the Master System. The primary purpose of
Namenode is to manage all the MetaData. Metadata is the list of files
stored in HDFS(Hadoop Distributed File System). As we know the data is
stored in the form of blocks in a Hadoop cluster. So the DataNode on
which or the location at which that block of the file is stored is mentioned
in MetaData. All information regarding the logs of the transactions
happening in a Hadoop cluster (when or who read/wrote the data) will be
stored in MetaData. MetaData is stored in the memory.

Features:
 It never stores the data that is present in the file.
 As Namenode works on the Master System, the Master system should
have good processing power and more RAM than Slaves.
 It stores the information of DataNode such as their Block id's and
Number of Blocks
How to start Name Node?
[Link] start namenode
How to stop Name Node?
[Link] stop namenode

2. DataNode
DataNode works on the Slave system. The NameNode always instructs
DataNode for storing the Data. DataNode is a program that runs on the
slave system that serves the read/write request from the client. As the
data is stored in this DataNode, they should possess high memory to
store more Data.

How to start Data Node?

[Link] start datanode
How to stop Data Node?
[Link] stop datanode

3. Secondary NameNode
Secondary NameNode is used for taking the hourly backup of the data. In
case the Hadoop cluster fails, or crashes, the secondary Namenode will
take the hourly backup or checkpoints of that data and store this data into
a file name fsimage. This file then gets transferred to a new system. A
new MetaData is assigned to that new system and a new Master is
created with this MetaData, and the cluster is made to run again
correctly.
This is the benefit of Secondary Name Node. Now in Hadoop2, we have
High-Availability and Federation features that minimize the importance of
this Secondary Name Node in Hadoop2.
Major Function Of Secondary NameNode:
 It groups the Edit logs and Fsimage from NameNode together.
 It continuously reads the MetaData from the RAM of NameNode and
writes into the Hard Disk.
As secondary NameNode keeps track of checkpoints in a Hadoop
Distributed File System, it is also known as the checkpoint Node.

The Hadoop Daemon's Port

Name Node 50070

Data Node 50075

Secondary Name Node 50090

These ports can be configured manually in [Link] and mapred-

[Link] files.
4. Resource Manager
Resource Manager is also known as the Global Master Daemon that
works on the Master System. The Resource Manager Manages the
resources for the applications that are running in a Hadoop Cluster. The
Resource Manager Mainly consists of 2 things.

A. ApplicationsManager
B. Scheduler
An Application Manager is responsible for accepting the request for a
client and also makes a memory resource on the Slaves in a Hadoop
cluster to host the Application Master. The scheduler is utilized for
providing resources for applications in a Hadoop cluster and for
monitoring this application.

How to start ResourceManager?

[Link] start resourcemanager

How to stop ResourceManager?
stop:[Link] stop resourcemanager

5. Node Manager
The Node Manager works on the Slaves System that manages the
memory resource within the Node and Memory Disk. Each Slave Node in
a Hadoop cluster has a single NodeManager Daemon running in it. It also
sends this monitoring information to the Resource Manager.

How to start Node Manager?

[Link] start nodemanager
How to stop Node Manager?
[Link] stop nodemanager

In a Hadoop cluster, Resource Manager and Node Manager can be

tracked with the specific URLs, of type [Link]
The Hadoop Daemon's Port

ResourceManager 8088

NodeManager 8042

The below diagram shows how Hadoop works.