BigData - Getting Started with HDFS

1.       Some basic facts about HDFS are , its distributed file system and build for batch processing with high fault tolerance.

2.       HDFS Build under the premise the hardward will fail. It has 1 NameNode and several DataNodes in a cluster. A cluster does not need to be in same premise, it can be distributed across regions over internet

3.       Diff between HDFS 1.0 and HDFS 2.0

a.       HDFS 1.0

                                                               i.      Single NameNode which is single point of failure

                                                             ii.      Resource manager and node manager are part of MapReduce

                                                           iii.      Scalability and Performance suffer with large cluster

b.       HDFS 2.0

                                                               i.      NameNode highly available with secondary name node and metadata files (fsimage and edit)

                                                             ii.      YARN was introduced for resource management and node management

                                                           iii.      Bringing in YARN helped fixing scalability and performance issues with large cluster

4.       Actions allowed in HDFS

a.       Copy files from local to HDFS and HDFS to Local

b.       Create New Files

c.       Append Files

d.       Read Files

e.       Delete Files

5.       Actions not allowed in HDFS

a.       Edit Files – HDFS does not allow file modifications

6.       Using HDFS Shell commands

a.       Add HDFS installed bin directory to PATH variable

b.       Then use “hdfs dfs -command [options] [input] [output]”

c.       We can also “hadoop fs -command [options] [input] [output]”

d.       If default file system is not hdfs then use Hadoop command , but with hdfs installed both are same

e.       Get all help using hdfs dfs -help [command]

f.        Basic commands

                                                               i.      Hdfs dfs -help

                                                             ii.      Hdfs dfs -ls

                                                           iii.      Hdfs dfs -touchz filename  (like linux touch command)

                                                           iv.      Hdfs dfs -cat filename

                                                             v.      Hdfs dfs -mkdir directoryname

                                                           vi.      Hdfs dfs -cp sourcefile targetfile (This copied files within HDFS)

                                                          vii.      Hdfs dfs -mv filename1 filename2

g.       File and Directory permissions

                                                               i.      Hdfs dfs -chmod [num] file name

                                                             ii.      Permisions are same as linux permissions

                                                           iii.      Other commands like -chgrp, -chown etc same as linux

h.       Moving files between HDFS and Local file system

                                                               i.      -put to put files insider HDFS

                                                             ii.      -get to get files from HDFS to local file systems

                                                           iii.      Also we can user -copyToLocal and -copyFromLocal

                                                           iv.      Another command is -moveFromLocal

i.         Maintenance Shell Commands

                                                               i.      Hdfs dfs -rm -r (to remove files and empty directory).. this does not really physically delete it.. rm is only trash it unlike in linux file system

                                                             ii.      Hdfs dfs -expunge .. is to clear the trash i.e. physically or permanently deleting the files from trash

7.       Transferring relational data to HDFS using Sqoop

a.       Sqoop is a tool to bulk transfer of data between relational database and HDFS

b.       Sqoop does this by running a MapReduce job and using some of HDFS basic commands.

c.       Sqoop has matured APIs for most of relational database.. most popular ones are MySQL, Oracle, SQL Server, PostgreSQL

d.       Sqoop does not offer out of box NoSQL connectors but there are lot of 3^rd party vendors offer NoSQL connectors in Sqoop

e.       Spoop is an Apache open source project

f.        Spoop allows Automation of workflows like importing and exporting data

g.       Spoop is two-way .. we can take read data and upload data in and out of both HDFS and RDMS

h.       Some of Sqoop user cases

                                                               i.      If we want to run big data analytics on transactional RDMS database without impacting online system.. we can automate regular data copy from RDMS to HDFS in Sqoop

                                                             ii.      If we have multiple silo databases and we want to run analytics on consolidated data, we can automate in sqoop to consolidate from multiple sources to single HDFS

i.         Basic sqoop commands

                                                               i.      Sqoop list-tables --connect jbdc:mysql://(MySQL Address)/dbname

                                                             ii.      Sqoop import --connect jdbc:mysql://(MySQL Address)/dbname --table tablename  -m 1 (-m option specifies number of MapReduce jobs to run)

                                                           iii.      Sqoop import --connect jdbc:mysql://(MySQL Address)/dbname --query select * from …..

                                                           iv.      For more information Sqoop.apache.org

8.       Querying data with Pig and Hive

a.       Hive – data warehouse software that works on top of Hadoop and allows for developers to structure data into schemas

                                                               i.      We can store both structured and un-structured data into HDFS. But Hive is a tool that deals with structured data, its schema bound. We need first move hdfs structured/unstructured data to Hive (schema bound) and use HiveQL to query

                                                             ii.      We can write HiveQL just like SQL on RDMS

                                                           iii.      Just type hive to start hive prompt

                                                           iv.      We can run hdfs commands also in hive prompt

                                                             v.      Querying HDFS using HiveQL is 3 step process

1.       Build data in Hive

a.       Hive> CREATE DATABASE dbname;

b.       Hive> USE database (to set database as default for querying)

c.       Hive>CREATE TABLE tablanme(field1 String, field2 INT, ….) row format fields delimited by “,”  lines delimited by ‘\n’ tblproperties(“skip.header.line.count”=”1”);

d.       Hive>LOAD DATA INPATH ‘/user../filename.csv’OVERWRITE INTO TABLE tablename;

2.       Write HiveQL query

a.       Select fields from tablename where conditions;

3.       Extract results

b.       Pig – Pig is the application environment used to run Pig Latin and covert Pig Latin scripts to MapReduce jobs

                                                               i.      Pig is not bound to schema

                                                             ii.      Pig can be used over unstructured or structured data

                                                           iii.      Pig interactive grunt shell can be started using “Pig -x local” for local file access and “Pig -x mapreduce” or just “Pig” for HDFS file access

                                                           iv.      Pig> help (will display all commands help)

                                                             v.      Pig interactive grunt shell is more interactive than Hive therefore we can run commands like “ls”, “cd” just as we type on shell interactive prompt

c.       Sample pig scripts

                                                               i.      Load data from HDFS

1.       A = LOAD ‘/user/../filename’ USING PigStorage(‘,’)  as (field1:String, fields2:INT….);

2.       Pig does not run the command until its called.. for example above A variable not evaluated until its selected

3.       When we do “DUMP A” then the MapReduce job runs to load hdfs filename to A variable in Pig

                                                             ii.      Write JOIN in Pig Latin

1.       A_new = FOR EACH new GENERATE field1,field2;

2.       A_combined=JOIN var1 BY field1, var2 by field1;

                                                           iii.      Store results in HDFS

1.       STORE A_combined /user/…/outputfilename USING PigStorage(‘,’);

2.       Hive vrs Pig

a.       Hive excels in data warehouse scenario. Hive uses declarative SQL like language. Its schema bound.

b.       It can be deal with unstructured or structured. Its fits best in ETL type transactions, which normally deal with semi-structured. Text heavy documents, Log Files are some example where Pig could be handy. It uses procedural Pig Latic scripting

c.       “SQL on Hadoop – Analyzing BigData with Hive” and “Pig Latin: Getting Started” courses

9.       HBASE

a.       HBASE is NoSQL distributed and scalable database built on top of Hadoop. Real-time read/write in Hadoop.

b.       HBASE can “Host data with billions of rows and millions of columns”

c.       HBASE works in highly distributed and scalable environment

d.       HBASE interactive shell can be started with command “hbase shell”

e.       Basic commands

                                                               i.      To create a table “user” with column “username” ---- “create ‘user’,’username’”

                                                             ii.      To select contents in “user” table.. just type “scan ‘user’”

                                                           iii.      To put row number 1 , columnd usename and value “Raj”.. the command will be .. hbase> put ‘user’,’r1’,’username’,’Raj’

                                                           iv.      To update ‘Raj’ to ‘Rajendra’ … hbase> put ‘user’,’r1’,’username’,’Rajendra’

                                                             v.      To drop the table .. disable ‘user’  then drop ‘user’

f.        We can user Pig to load/unload HDFS files into/from HBASE

10.   We can user Bash shell scripting to automate HDFS files managing tasks.

BigData - MapReduce programming concepts

1.       Map logic is coded in a class that extends the standard mapper class

2.       Map class take Input <key, value> and produces output <Key, value>. The input <key, value> is determined by how the input file on disk is read.  Typically input files are text files. The Mapper framework reads input text file line by line and the line number is assigned to Key and line itself is assigned value field. Output key,value depends on how have you processed input key,value pairs.

3.       Reduce logic is coded in a class that extends Reducer class. The output key,value of map class should match with reduce input key,value.

4.       Main class is driver program. This is class that set up mapreduce job. In this job we need to specify map class and reduce class the  we need to specific input file path and output file path etc..

5.       Datatypes

a.       LongWritable is Hadoop wrapper for Long datatype. It implements writable interface provided by the Hadoop library. The writable interface allows to serialize and deserialize data from storage system. All keys should be writable to disk and readable from disk which is why key always have to be writable classes.

b.       Text is a String type. It’s a Hadoop is writable class which wraps Java String. Text data type allows easily interoperate with other tools which understand UTF8.

c.       IntWritable is Hadoop wrapper for Int just like other Hadoop wrappers.

6.       Within map class, we will have to overwrite map method

a.       Map method take 3 parameters. Input Key,value and context

                                                               i.      Input key and value are passed from the class parameters

                                                             ii.      Context is where actual internal shuffle and sort logic happen before passing data to reducer class.

                                                           iii.      The output of map methods should be written to context using context.write(key,value) in order to pass to reducer class

7.       The map reduce frameworks behind the scenes to collect all  map class output and finds all values with same key and collects them together , and offer this as collection input to reduce class (this is what happens in context.. shuffle and sort)

8.       The reduce class input should match with output of map class. The reduce class takes key and value input but the value will be collection …. This is because map reduce framework collected all map class output and produced key and collections of values by sort and shuffle. Therefore the value column should be defined as iterable for reduce class

9.        Within reduce class, we will have to overwrite reduce method

a.       Reduce method takes 3 parameters. Input key, Interable(Value), Context

                                                               i.      Context is same map class. This context is an objects provides bridge to the external framework.. in simple works it stores final output that mapreduce job should produce. So at then end of reduce method, we need to write to context.write() to write to mapreduce job output

10.   The Main class extends Configured class and implements tool interface

a.       Configured class is base class for any application in Hadoop framework which has configurable parameters

b.        Tool Interface essentially says any application run in Hadoop environment will pick up right command line arguments that it needs

c.       Tool along with ToolRunner class accepts application specific command line arguments

d.       We will have to overwrite run method .. run method is from Tool interface

                                                               i.      In run method, the mapreduce job is configured and instantiated

                                                             ii.      ToolRunner class extract command line arguments and sets the mapreduce job  running

11.   In terms of parallel processing

a.       Number of map tasks are determined by number file partitions which is determined by underlying file system. This is not configurable and cannot be altered

b.       We can configure number of reduce tasks. Default is 1. This can be set in the main class using job.setNumReduceTasks(no);

c.       If we configure more than 1 reduce tasks, the partition job internal to mapreduce frameworks assigns each key of map output to a partition and each partition will be processed by each reduce task

d.       The internal partition jobs ensures that all same keys goes to same partition. i.e. same can not be sent to multiple reducers, this is ensured by partitioned internal jobs

e.       Finally all reduce tasks output is combined to give final mapreduce output

12.   To improve MAP class performance, we can create Combiner

a.       Combiner does the reduce task that can be performed on individual nodes

b.       For example, if we are calculating sum.. combiner can perform sum on mad node output and pass it to reducer. Which mean reducers will have to do less

c.       To set the combiner class , all we have to do is setting a parameter in Main class.. job.setCombinerClass(Reduce.class)

d.       Combiners will increase the parallelism and reduce network data transfer

e.       Most of the cases , we can user reduce class as combiner. But some cases like AVG , we can not user reduce class as combiner therefore it has be to different class.

BigData - Working with HDFS, MapReduce and YARN

1.       HDFS -  Hadoop Distributed File System

a.       HDFS is default file system used in Hadoop

b.       As name suggests, HDFS spread out multiple nodes where each node is individual machine built on commodity hardware

c.       HDFS is highly fault tolerant. Individual nodes might go down or have corrupted data but it does no impact availability of data in cluster as whole.

d.       Its suited for batch processing with data access high throughput rather than low latency. This means it is suited for large volumes and long running jobs which process data for long time. Its not suited for quick retrieval of data

e.       Queries to HDFS are not interactive  i.e. user can not type SQL and wait for response . all queries are via batch jobs

f.        HDFS used to store very very large files

g.       HDFS data is semi-structured data unlike relational database.  It tends to be unformatted and unstructured

h.       HDFS files are stored on multiple disks where each disk is on different machines in cluster. HDFS file system responsibility is to manage all these disks. It does by configuring master-slave 

i.          Within cluster HDFS chooses randomly one node as masternode.. masternode is responsible for coordinating storage all other slave nodes.

j.         On masternode, HDFS runs a java process which receives all requests that are made by user on cluster and forwards these requests to other slavenodes, this process is called NameNode.

k.       HDFS runs a process called DataNode in all other slave nodes.

l.         In simple terms, master is NameNode and slave is DataNode.. Only one NameNode per cluster. NameNode maintains information about the HDFS files and keeps record of what file is present in which datanode etc.. NameNode is single point for failure. I.e. if NameNode fails that’s it , HDFS is failed. As a remedy, we will have secondary NameNode as a fail over.

m.      Storing data in HDFS

                                                               i.      Each large file is divided into equal blocks. Each block or group of blocks are stored in different nodes

                                                             ii.      NameNode maintains records of which block stored in which DataNode

                                                           iii.      Multiple same copies of blocks are stored in different nodes for fault tolerance

                                                           iv.      Optimum block size in HDFS is 128mb. If we increase blocksize, parallelism is reduced. If we decrease the blocksize, it will increase the overhead.

                                                             v.      Time taken to process a block can be divided into 2 parts.. 1 to locate the block (seek time) and 2 to read the block (transfer time).

n.       HDFS Commands

                                                               i.      Hdfs file systems commands are same as linux file system commands example mkdir, rmdir, ls etc

                                                             ii.      From Hadoop installation directory , run “bin/hadoop fs -help” to get man page on commands.. we can also use “bin/hdfs dfs -help”…the difference is Hadoop command can deal with other file system configured in Hadoop but hdfs only deals with hdfs file system

                                                           iii.      We can copy/move files between local file system and Hadoop hdfs file system

1.       Bin/hdfs dfs -copyFromLocal local-file hdfs-file…  for copying file from local to hdfs… -put command on hdfs also does same thing

2.       Bin/hdfs dfs -copyToLocal hdfs-file local-file  -- for copying file from hdfs to local..  -get command on hdfs also does same thing

o.       HDFS Fault Tolerance is achieved using replications

                                                               i.      Default replication factor in fully distributed environment is 3.. i.e. every block is duplicated 3 times

                                                             ii.      Default replication factor in pseudo distrusted model is 1

                                                           iii.      Replication is done by balancing maximum redundancy and minimum write bandwidth

1.       maximum redundancy is achieved by replicating in faraway nodes, typically in different racks in a cluster. i.e. even if entire rack goes down, we still have a copy of block to work with

2.       minimum write bandwidth is achieved by keeping replications as close as possible to avoid write latency at the time of replication

3.       minimum write bandwidth could be achieved by keeping replications in same rack in cluster because intra rack(within rack) is low latency compared with inter rack(between racks)…  but this also means low fault tolerance therefore balance needs to be maintained between maximum redundancy and minimum write bandwidth

p.       NameNode Failure Management

                                                               i.      NameNode stores all block mapping information in memory its not persistant. This means if NameNode fails then this information is lost

                                                             ii.      Every time NameNode is restarted, datanodes sends the block storage information to NameNode so that NameNode can reconstruct the block mapping information

                                                           iii.      Restarting NameNode is not ideal solution in production environements

                                                           iv.      Therefore in order to manage NameNode failures, we can use MetaData files and seconday name node

1.       There are 2 special metadata files which allows us reconstruct the name nodes

a.       fsimage  and edit files

b.       fsimage file holds snapshot of entire file system when Hadoop cluster started up

c.       edit file contains the in memory log information about all the in memory edits to file system

d.       fsimage and edit files together will give current state of file system

e.       Backup location of fsimage and edits will be outside HDFS, either local or remote file system.. this is set in hdfs-site.xml with property name dfs.namenode.name.dir

2.       Merging fsimage and edit files is a heavy process and long running process

3.       Other easier way to set up the secondary NameNode. Exact backup of complete original NameNode and runs on different machine. Secondary NameNode copies over fsimage and edit files, then applies edits from edit file to fsimage this is called checkpointing. Then this checkpointed fsimage is copied back to primary NameNode and this process continues. Checkpoint frequency of this kind is configurable in hdfs-site.xml. the property name is dfs.namenode.checkpoint.period.

2.       Processing data with MapReduce

a.       Processing huge data requires running processes on many machines. These machines work together as a distributed system. MapReduce is a programming paradigm on a distributed system.

b.       MAP process works on individual machines and work on data that live on its individual machine. REDUCE process collates output from individual MAP processes and transfers across network within cluster to one machine where reduce phase runs.

c.       We should write code in MAP process to work on one record and produce output in key-value pairs

d.       Reduce phase collates all key-value pairs and aggregates these key-value pairs into one final output

e.       Example.. if we want to find word frequencies in very very large text files

                                                               i.      Text file with lines and sentences.. its distributed across multiple nodes in cluster.. i.e. each node has a partition of a text file

                                                             ii.      MAP process runs on each machine and process partition on that machine and procedure key value pair.. in case of work frequencies, out put will be word and count of work example {word, count}.. node1: {twinkle, 2},{this,1} and node2: {twinkle, 1}, {this,2}

                                                           iii.      Then this MAP out will be passed to Reduce and Reduce will sum up count values per word and then produce final output.. i.e. {twinkle 3}, {this 3}

f.        MapReduce typically implemented in JAVA

                                                               i.      Every MapReduce comprises of 3 basic components i.e. MAP, REDUCE and MAIN

                                                             ii.      MAP Class is where map logic is implemented. REDUCE class is where reduce logic is implemented. MAIN is a driver program that sets up the job

                                                           iii.      MAP class extends the Mapper class and the method that we need to overwrite is MAP… map class is generic class it has 4 type parameters these are input key, input value, output key and output value

                                                           iv.      REDUCE classes extends reducer class and the method that we need to overwrite is REDUCE Method. Reduce class is generic class it has 4 type parameters these are input key, input value, output key and output value.. the input key,value types of reduce class should match with output key,value type

                                                             v.      MAIN class set ups the job instance which contains the configuration parameters of mapreduce and serves as entry point for our mapreduce job. The job instance is submitted to Hadoop cluster in order to get mapreduce running.

                                                           vi.      The jobs objects has properties to be configured.. some of them are Input Filepath, Output Filepath, Mapper Class, Reduce Class, Output Data types

3.       YARN

a.       YARN is introduced in 2013 as part of Hadoop 2.0. Before 2.0, YARN functions were performed as part of MapReduce. In 2013, MapReduce is divided into 2 parts i.e. MapReduce and YARN

b.       YARN has 2 deamons… resource manager that runs on NameNode and NodeManager that runs on all datanodes

c.       Resource manager manages resources like, disk space , memory, cpu etc across all nodes in the cluster

d.       NodeManager only looks at the node its running on. Its responsible for scheduling the tasks which runs just on that node and it communicates with the resource manager

e.       YARN has following scheduling polices

                                                               i.      FIFO – First In First Out .. its rarely used in YARN and it results in long waiting

                                                             ii.      Capacity Scheduler (This is default) – Its splits up the resources into queues and each queue will have a share of resources.. percentage of share is configurable based on importance of the queue…. Each queue can be categorized into functional subject area, we can submit the job into a queue.. within queue, its again FIFO scheduling… this scheduling can lead to under utilization of resources in cluster.

                                                           iii.       Faire Scheduling – Resources in cluster are always allocated to all jobs fairly … this means jobs submitted will immediately start running and resources are split across jobs.

                                                           iv.      Actual scheduling policy to be used can be set in yarn-site.xml .. the property name is yarn.resourcemenager.scheduler.class

                                                             v.      For Capacity Scheduling, the Queues can be set up in etc/hadoop/capacity-scheduler.xml

                                                           vi.      While submitting MapReduce job, we can use “-D mapreduce.job.queuename=<qname>”