Introduction to Hadoop

In 1990’s around 1400 MB of data was the maximum and had  a transfer speed of 4.4 MB/s, i.e. It would take around 311 sec or almost 5 minutes to transfer data. 20 years later, one terabyte drives are the norm, but the transfer speed is around 100 MB/S. Time required to read the whole disk is (1*1024*1024/(100*60*60)) = 2.9 hours.

The obvious way to reduce the time is to read from multiple disks at once. Imagine if we had 100 drives, each holding one hundredth of the data. Working in parallel, we could read the data in under two minutes.  ==>2.9*100/60 .The first problem to solve is hardware failure: as soon as you start using many pieces of hardware, the chance that one will fail is fairly high. A common way of avoiding data loss is through replication: redundant copies of the data are kept by the system so that in the event of failure, there is another copy available. This is how RAID works, for instance, although Hadoop’s file system, the Hadoop Distributed File system (HDFS), takes a slightly different approach.

The second problem is that most analysis tasks need to be able to combine the data in some way; data read from one disk may need to be combined with the data from any of the other 99 disks. Hadoop provides: a reliable shared storage and analysis system. The storage is provided by HDFS and analysis by Map Reduce.

Seeking:  is the process of moving the disk’s head to a particular place on the disk to read or write data. It characterizes the latency of a disk operation, whereas the transfer rate corresponds to a disk’s bandwidth.

Map Reduce VS RDBMS:

Map Reduce is a good fit for problems that need to analyze the whole data set, in a batch fashion, particularly for ad hoc analysis. An RDBMS is good for point queries or updates, where the data set has been indexed to deliver low-latency retrieval and update times of a relatively small amount of data.

Map Reduce:  Suits applications where the data is written once, and read many times, relational database is good for data sets that are continually updated. Relational data is often normalized to retain its integrity and remove redundancy. Normalization poses problems for Map Reduce. A web server log is a good example of a set of records that is not normalized.

If you double the size of the input data, a job will run twice as slow. But if you also double the size of the cluster, a job will run as fast as the original one. This is not generally true of SQL queries. Map Reduce tries to collocate the data with the compute node, so data access is fast since it is local. This feature, known as data locality, is at the heart of Map Reduce and is the reason for its good performance.

Map Reduce are limited to key and value types that are related in specified ways, and mappers and reducers run with very limited coordination between one another (the mappers pass keys and values to reducers).

Hadoop introduction:

Created by Doug Cutting In 2004, Google published the paper that introduced Map Reduce to the world.

Nutch developers had a working Map Reduce implementation in Nutch(Apache) 

Hadoop & Hadoop Ecosystem:

Common: A set of components and interfaces for distributed file systems and general I/O

(Serialization, Java RPC, persistent data structures).

• HDFS: Distributed file system that runs on large clusters of commodity machines
• Pig:  Data flow language, explores very large datasets.
• Hive: Distributed data warehouse
• HBase: Column-oriented database
• Zookeeper:  Distributed, highly available coordination service.
• Sqoop: Moving data between relational databases and HDFS.

Map Reduce:

• Programming model for data processing
• Map Reduce programs are inherently parallel.
• What’s the highest recorded global temperature for each year in the dataset?
• Performance baseline

Map Reduce works by breaking the processing into two phases: the map phase and the reduce phase. Each phase has key-value pairs as input and output, the types of which may be chosen by the programmer. The programmer also specifies two functions: the map function and the reduce function.

Choose text input format each line can be a text values, & key be the offset.

Use case find max temperature

Sample data:

0067011990999991950051507004...9999999N9+00001+99999999999...
0043011990999991950051512004...9999999N9+00221+99999999999...
0043011990999991950051518004...9999999N9-00111+99999999999...
0043012650999991949032412004...0500001N9+01111+99999999999...
0043012650999991949032418004...0500001N9+00781+99999999999...

Data presented to map function (Key -Value pair)

(0, 0067011990999991950051507004...9999999N9+00001+99999999999...)
(106, 0043011990999991950051512004...9999999N9+00221+99999999999...)
(212, 0043011990999991950051518004...9999999N9-00111+99999999999...)
(318, 0043012650999991949032412004...0500001N9+01111+99999999999...)
(424, 0043012650999991949032418004...0500001N9+00781+99999999999...)

Map function output

(1950, 0)
(1950, 22)
(1950, −11)
(1949, 111)
(1949, 78)

The output from the map function is processed by the Map Reduce framework before

Being sent to the reduce function. =>Sorts & Groups [Combiner]

(1949, [111, 78])
(1950, [0, 22, −11])

Reducer:

All the reduce function has to do now is iterate through the list and pick up the maximum reading:

(1949, 111)
(1950, 22)

Java Map Reduce Code:

3 things =>Map function, Reduce function, some code to run job.

Creating project:

Create a new java project

Add jar from: Hadoop home & Hadoop Home/lib

C:\Hadoop\hadoop-1.1.0-SNAPSHOT

C:\Hadoop\hadoop-1.1.0-SNAPSHOT\lib

General procedure:

Set a job

Assign mapper class

Assign reducer class

CustomMapper class extends Mapper (input key, input value, outputKey, outputValue)

Input key -- some key

Input value --Line

Output Key -- year

Output Value --MaxTemp

map function (inputKey, inputValue)

Hadoop own data types:

These are found in the org.apache.hadoop.io package.

LongWritable ==>Like  Long

Text  =>Like String

IntWritable ==>Like Integer

Setting jar by class:

Hadoop will distribute this file around the cluster, & locate the relevant jar.

job.setJarByClass(MaxTemperatureDriver.class);

Specify input and output path:

FileInputFormat.addInputPath(job, new Path(args[0]));

FileOutputFormat.setOutputPath(job,new Path(args[1]));

Setting mapper and reducers:

Hadoop send individual jar to each

job.setMapperClass(MaxTemperatureMapper.class);

job.setReducerClass(MaxTemperatureReducer.class);

Waiting for completion:

After setting the job, hadoop will wait for completion of job. This is done by :

(job.waitForCompletion(true) ? 0:1)

Data Flow:

Hadoop runs the job by dividing it into tasks:  map tasks and reduce tasks.

Job Execution process is controlled by 2 types of nodes:

• A job tracker : coordinates all the jobs run on the system by scheduling tasks to run on tasktrackers.
• Task trackers: run tasks and send progress reports to the job tracker. Task fails, the job tracker can reschedule it on a different task tracker.

Splits:

• Input to Map reduce job are split into fixed pieces by Hadoop.
• Creates one map task for each split.
• Many splits, less time to process rather than whole input.
• Processing splits in parallel will make it better load balanced.
• Faster machine will be able to process it more proportionally.
• Splits are too small managing the splits and of map task creation begins to dominate the total job execution time.
• Good split size tends to be the size of an HDFS block, 64MB by default.

Data locality optimization:

• Run the map task on a node where the input data resides in HDFS.
• Helps reduce cluster b/w
• Splits can be spanned over multiple blocks in HDFS.

Map tasks & Reduce task:

• As maps tasks are intermediate steps, those are stored in local disk instead of HDFS because it is processed by reduced task to produce the final output.
• The input to the single reducer is normally the output from the all the mappers.
• The output of reduce are tasks are stored in HDFS.
• The first replica stored at local node & other replicas are stored at off rack node.

Combiner:

• To minimize the data transferred between map and reduce tasks, Hadoop uses combiners after mapper to combine multiple mappers output which is then send to reducer.
• After setting mapper class, set combiner if you need by and defined on Reducer class :

job.setCombinerClass(MaxTemperatureReducer.class);

Reducer:

• Hadoop Streaming uses UNIX standard streams as the interface between Hadoop and your program.
• Streaming is naturally suited for text processing
• Hadoop Pipes: C++ interface to Hadoop Map Reduce

Configuration:

C:\hadoop\hadoop-1.1.0-SNAPSHOT\conf

hdfs-site.xml:

dfs.replication ⇒ This property set the replication factor for data.

dfs.webhdfs.enabled

Summary:

Active Hadoop node || Passive Hadoop (Secondary name node)

Name node server: is a Hadoop name node server that runs

20-35 Peta Bytes of data large cluster can support.

Underlying file System options [HDFS]:

You do not format the hard drive with HDFS but use ext3, ext4 or XFS slave hard drive.

HDFS =>Abstract file system

Ext4 recommendations for hard drive format.

Commands:

hadoop fs -help

bin

/hadoop fs

-ls

-mkdir

-copyFromLocal

-copyToLocal

-moveToLocal

-rm

-tail

-chmod

-setrep -w 4 -R  /dir1/s-dir/   //Replication factor change to 4 recursively for this subdirectory

Yahoo 4500 node clusters:

110 Racks

Each rack 40 slaves

At top there is rack switch for each rack

8 core switches

Every slaves machine has 2 Cat five cables  going at top of rack switch, that means each to of rack switch has 40*2 = 80 Ports in each rack.

Within rack its’ 1GB network, However core switch layer is 10GB network, of which 8GB           is dedicated for HDFS & rest 2GB for Map reduce administration, & user traffic on the network.

Rack awareness:

 Name node is rack aware.

Commands:

Hadoop

     FS

What are the files at root of HDFS?

haddop fs -ls /

dfs -ls /user/hduser/  ==>works at both linux & windows

Make new dir:

mkdir  /user/clusdra/newDir

hadoop fs -ls copyFromLocaal shakesspear.txt /user/username/newDir

Filesystem check command:

% hadoop fsck / -files -blocks

in windows: hadoop fsck \ -files -blocks

-copyToLocal:

• hadoop fs -copyToLocal hdfs://localhost/user/hadoop/test1.txt C:\Users\prems.bist\Desktop\test.txt

-copyFromLocal:

hadoop fs -copyFromLocal D:\tutorials\test.txt hdfs://localhost/user/hadoop/test1.txt

Listing files inside the directory of hdfs:

hadoop fs -ls hdfs://localhost/user/hadoop/yourFolder => Enlists all the files at yourFolder dir.

can use the relative path as well as:

hadoop fs -copyFromLocal input/docs/quangle.txt  /user/tom/quangle.txt

command Examples in windows:

hadoop fs -mkdir books

 //Creates directory book at C:\users\prems.bist  but it will not be seen thorough user                                                              prems.bist

hadoop fs -ls

Example 2 -command :

hadoop fs -mkdir input               //makes directory named input

hadoop fs -ls input                     //It will so nothing as there is no files

hadoop fs -put *.xml input       //Puts all the files starting with .xml  in input directory

output:

Found 1 items

drwxr-xr-x   - prems.bist supergroup          0 2014-03-13 13:09 /user/prems.bist/books

Column first -->filePermssion

Second column =>replication factory which is (-) thats means not defined yet.

Third column : Owner

Fourth column : group

Fifth column: size of the file in byte 0 for dierctory

6th & seventh column: Last modified date & time

8th Column: Absolutely name of file or directory

Examples 3 - command Running jar through command line:

Create directory

hadoop dfs -mkdir hdfs:/inputFolder

Show directory

hadoop dfs -ls hdfs:/

Copy the files from localSystem

hadoop dfs -copyFromLocal D:\SampleData\input hdfs:/inputFolder

Run the Jar file

hadoop jar jarName runningClassName inputFileLocation outputFileLocation

ex-

hadoop jar D:\target\MapReduce-1.0-SNAPSHOT.jar com.impetus.hadoop.WordCount hdfs:/wordCountFolder/sampleData hdfs:/outputFolder

Viewing file output:

hadoop dfs -cat /outputFolder/part-00000

List all the files in root directory of local filesystem ie( files of c drive):

hadoop fs -ls file:///  

HDFS:

Distributed filesystems:

• Filesystems that manage the storage across a network of machines.

Commodity hardware: Normal hardware devices

Streaming data access: Time to read all dataset is more important than reading a first one.

Low-latency data access:

Lots of small files: Since the namenode holds filesystem metadata in memory, the limit to the number of files in a filesystem is governed by the amount of memory on the namenode.

Multiple writers, arbitrary file modifications : Files in HDFS may be written to by a single writer

Block:

• Is the minimum amount of data that it a disk can read or write.
• File system blocks are typically a few kilobytes in size, while disk blocks are normally 512 bytes.
• HDFS, too, has the concept of a block, but it is a much larger unit—64 MB. Like in a file system for a single disk, files in HDFS are broken into block-sized chunks,
• Which are stored as independent units. HDFS blocks are large compared to disk blocks, and the reason is to minimize the cost of seeks.
• A quick calculation shows that if the seek time is around 10 ms, and the transfer rate is 100 MB/s, then to make the seek time 1% of the transfer time, we need to make the block size around 100 MB.
• +

Fail over and Fencing:

The transition from the active namenode to the standby is managed by a new entity in

The system called the failover controller.  The first implementation uses ZooKeeper to ensure that only one name node is active. Failover may also be initiated manually by an administrator, in the case of routine maintenance, for example. This is known as a graceful failover, since the failover controller

Arranges an orderly transition for both name nodes to switch roles. STONITH =>Shoot the other node in the head.

Default HDFS port:  8020

org.apache.hadoop.fs.FileSystem represents a file system in Hadoop.

Accessing HDFS over Http:

2 ways:

1) Using HDFS daemons that server http request to clients

2) Using distributed file system API’s (Using proxies)

Hadoop file systems:

Name	URI scheme	Class	Description
Local	file	fs.LocalFileSystem
HDFS	hdfs	hdfs.DistributedFileSystem
HFTP	hftp	hdfs.HftpFileSystem	Provides read only access to HDFS over http
HSFTP	hsftp	hdfs.HsftpFileSystem	Providing read-only access to HDFS over HTTPS.
WebHDFS	webhdfs	hdfs.web.WebHdfsFile System	Secured read write access to http.
HAR	har	fs.HarFileSystem	Archiving
KFS	kfs	fs.kfs.KosmosFileSystem	CloudStore writtenn in C++
FTP	ftp	fs.ftp.FTPFileSystem	FTP server backed file system
S3	s3		Amazon  backed
Distributed RAID	hdfs	hdfs.DistributedRaidFi leSystem	“RAID” version of HDFS
View	viewfs	viewfs.ViewFileSystem	client-side mount table for other Hadoop filesystems

File System functions:

FilSystem dfs = FileSystem.get(config);

getWorkingDirectory => Returns working path

delete ⇒ To delete file.

createNewFile()

exists(src)

getDefaultBlockSize()

available() //returns estimated number of byte remaining

//Create directory

hadoop dfs -mkdir hdfs:/inputFolder

//Show directory

fs:/ //Copy the files

hadoop dfs -ls h

d from localSystem

mLocal D:\SampleData\input hdfs:/inputFolder //Run the Jar file

hadoop dfs -copyFr

o

hadoop jar jarName runningClassName inputFileLocation outputFileLocation

hdfs:/wordCountFolder/sampleData hdfs:/outputFolder //output Can be seen as : hadoop dfs -cat /outputFolder/part-00000

hadoop jar D:\target\MapReduce-1.0-SNAPSHOT.jar com.impetus.hadoop.WordCount

References: Hadoop Defnitive Guide

Happy Learning cheers.!!