Big Insights with Big Data: March 2013

Monday, March 18, 2013

MongoDB Tech Talk

Please check out FREE training resources for MongoDB:
https://education.10gen.com/

Hadoop Qs for certification

1. What is a JobTracker in Hadoop? How many instances of JobTracker run on a Hadoop Cluster?

JobTracker is the daemon service for submitting and tracking MapReduce jobs in Hadoop. There is only One Job Tracker process run on any hadoop cluster. Job Tracker runs on its own JVM process. In a typical production cluster its run on a separate machine. Each slave node is configured with job tracker node location. The JobTracker is single point of failure for the Hadoop MapReduce service. If it goes down, all running jobs are halted. JobTracker in Hadoop performs following actions: (from Hadoop Wiki)

§ Client applications submit jobs to the Job tracker.

§ The JobTracker talks to the NameNode to determine the location of the data

§ The JobTracker locates TaskTracker nodes with available slots at or near the data

§ The JobTracker submits the work to the chosen TaskTracker nodes.

§ The TaskTracker nodes are monitored. If they do not submit heartbeat signals often enough, they are deemed to have failed and the work is scheduled on a different TaskTracker.

§ A TaskTracker will notify the JobTracker when a task fails. The JobTracker decides what to do then: it may resubmit the job elsewhere, it may mark that specific record as something to avoid, and it may even blacklist the TaskTracker as unreliable.

§ When the work is completed, the JobTracker updates its status.

§ Client applications can poll the JobTracker for information.

2. How JobTracker schedules a task?

The TaskTracker send out heartbeat messages to the JobTracker, usually every few minutes, to reassure the JobTracker that it is still alive. These messages also inform the JobTracker of the number of available slots, so the JobTracker can stay up to date with where in the cluster work can be delegated. When the JobTracker tries to find somewhere to schedule a task within the MapReduce operations, it first looks for an empty slot on the same server that hosts the DataNode containing the data, and if not, it looks for an empty slot on a machine in the same rack.

3. What is a Task Tracker in Hadoop? How many instances of TaskTracker run on a Hadoop Cluster

A TaskTracker is a slave node daemon in the cluster that accepts tasks (Map, Reduce and Shuffle operations) from a JobTracker. There is only One Task Tracker process run on any hadoop slave node. Task Tracker runs on its own JVM process. Every TaskTracker is configured with a set of slots. These indicate the number of tasks that it can accept. The TaskTracker starts a separate JVM processes to do the actual work (called as Task Instance) this is to ensure that process failure does not take down the task tracker. The TaskTracker monitors these task instances, capturing the output and exit codes. When the Task instances finish, successfully or not, the task tracker notifies the JobTracker. The TaskTrackers also send out heartbeat messages to the JobTracker, usually every few minutes, to reassure the JobTracker that it is still alive. These messages also inform the JobTracker of the number of available slots, so the JobTracker can stay up to date with where in the cluster work can be delegated.

4. What is a Task instance in Hadoop? Where does it run?

Task instances are the actual MapReduce jobs which are run on each slave node. The TaskTracker starts a separate JVM processes to do the actual work (called as Task Instance) this is to ensure that process failure does not take down the task tracker. Each Task Instance runs on its own JVM process. There can be multiple processes of task instance running on a slave node. This is based on the number of slots configured on task tracker. By default a new task instance JVM process is spawned for a task.

5. How many Daemon processes run on a Hadoop system?

Hadoop is comprised of five separate daemons. Each of these daemons runs in its own JVM.

Following 3 Daemons run on Master nodes

§ NameNode - This daemon stores and maintains the metadata for HDFS.

§ Secondary NameNode - Performs housekeeping functions for the NameNode.

§ JobTracker - Manages MapReduce jobs, distributes individual tasks to machines running the Task Tracker.

Following 2 Daemons run on each Slave nodes

§ DataNode – Stores actual HDFS data blocks.

§ TaskTracker - Responsible for instantiating and monitoring individual Map and Reduce tasks.

6. What is configuration of a typical slave node on Hadoop cluster? How many JVMs run on a slave node?

§ Single instance of a Task Tracker is run on each Slave node. Task tracker is run as a separate JVM process.

§ Single instance of a DataNode daemon is run on each Slave node. DataNode daemon is run as a separate JVM process.

§ One or Multiple instances of Task Instance is run on each slave node. Each task instance is run as a separate JVM process. The number of Task instances can be controlled by configuration. Typically a high end machine is configured to run more task instances.

7. What is the difference between HDFS and NAS?

The Hadoop Distributed File System (HDFS) is a distributed file system designed to run on commodity hardware. It has many similarities with existing distributed file systems. However, the differences from other distributed file systems are significant. Following are differences between HDFS and NAS

In HDFS Data Blocks are distributed across local drives of all machines in a cluster. Whereas in NAS data is stored on dedicated hardware.
HDFS is designed to work with MapReduce System, since computations are moved to data. NAS is not suitable for MapReduce since data is stored separately from the computations.

§ HDFS runs on a cluster of machines and provides redundancy using a replication protocol. Whereas NAS is provided by a single machine therefore does not provide data redundancy.

8. How NameNode Handles data node failures?

NameNode periodically receives a Heartbeat and a Blockreport from each of the DataNodes in the cluster. Receipt of a Heartbeat implies that the DataNode is functioning properly. A Blockreport contains a list of all blocks on a DataNode. When NameNode notices that it has not received a heartbeat message from a data node after a certain amount of time, the data node is marked as dead. Since blocks will be under replicated the system begins replicating the blocks that were stored on the dead datanode. The NameNode Orchestrates the replication of data blocks from one datanode to another. The replication data transfer happens directly between datanodes and the data never passes through the namenode.

9. Does MapReduce programming model provide a way for reducers to communicate with each other? In a MapReduce job can a reducer communicate with another reducer?

Nope, MapReduce programming model does not allow reducers to communicate with each other. Reducers run in isolation.

10. Can I set the number of reducers to zero?

Yes, setting the number of reducers to zero is a valid configuration in Hadoop. When you set the reducers to zero no reducers will be executed, and the output of each mapper will be stored to a separate file on HDFS. [This is different from the condition when reducers are set to a number greater than zero and the Mappers output (intermediate data) is written to the Local file system (NOT HDFS) of each mapper slave node.]

11. Where is the Mapper Output (intermediate key-value data) stored?

The mapper output (intermediate data) is stored on the Local file system (NOT HDFS) of each individual mapper nodes. This is typically a temporary directory location which can be setup in config by the hadoop administrator. The intermediate data is cleaned up after the Hadoop Job completes.

12. What are combiners? When should I use a combiner in my MapReduce Job? Combiners are used to increase the efficiency of a MapReduce program. They are used to aggregate intermediate map output locally on individual mapper outputs. Combiners can help you reduce the amount of data that needs to be transferred across to the reducers. You can use your reducer code as a combiner if the operation performed is commutative and associative. The execution of combiner is not guaranteed. Hadoop may or may not execute a combiner. Also, if required it may execute it more than 1 times. Therefore your MapReduce jobs should not depend on the combiner’s execution.

13. What is Writable & WritableComparable interface?

§ org.apache.hadoop.io.Writable is a Java interface. Any key or value type in the Hadoop Map-Reduce framework implements this interface. Implementations typically implement a static read (DataInput) method which constructs a new instance, calls readFields (DataInput) and returns the instance.

§ org.apache.hadoop.io.WritableComparable is a Java interface. Any type which is to be used as a key in the Hadoop Map-Reduce framework should implement this interface. WritableComparable objects can be compared to each other using Comparators.

14. What is the Hadoop MapReduce API contract for a key and value Class?

§ The Key must implement the org.apache.hadoop.io.WritableComparable interface.

§ The value must implement the org.apache.hadoop.io.Writable interface.

15. What are IdentityMapper and IdentityReducer in MapReduce?

§ org.apache.hadoop.mapred.lib.IdentityMapper Implements the identity function, mapping inputs directly to outputs. If MapReduce programmers do not set the Mapper Class using JobConf.setMapperClass then IdentityMapper.class is used as a default value.

§ org.apache.hadoop.mapred.lib.IdentityReducer Performs no reduction, writing all input values directly to the output. If MapReduce programmers do not set the Reducer Class using JobConf.setReducerClass then IdentityReducer.class is used as a default value.

16. What is the meaning of speculative execution in Hadoop? Why is it important? Speculative execution is a way of coping with individual Machine performance. In large clusters where hundreds or thousands of machines are involved there may be machines which are not performing as fast as others. This may result in delays in a full job due to only one machine not performing well. To avoid this, speculative execution in hadoop can run multiple copies of same map or reduce task on different slave nodes. The results from first node to finish are used.

17. When are the reducers started in a MapReduce job?

In a MapReduce job reducers do not start executing the reduce method until the all Map jobs have completed. Reducers start copying intermediate key-value pairs from the mappers as soon as they are available. The programmer defined reduce method is called only after all the mappers have finished.

18. If reducers do not start before all mappers finish then why does the progress on MapReduce job shows something like Map (50%) Reduce (10%)? Why reducer’s progress percentage is displayed when mapper is not finished yet?

Reducers start copying intermediate key-value pairs from the mappers as soon as they are available. The progress calculation also takes in account the processing of data transfer which is done by reduce process, therefore the reduce progress starts showing up as soon as any intermediate key-value pair for a mapper is available to be transferred to reducer. Though the reducer progress is updated still the programmer defined reduce method is called only after all the mappers have finished.

19. What is HDFS? How it is different from traditional file systems?

HDFS, the Hadoop Distributed File System, is responsible for storing huge data on the cluster. This is a distributed file system designed to run on commodity hardware. It has many similarities with existing distributed file systems. However, the differences from other distributed file systems are significant.

§ HDFS is highly fault-tolerant and is designed to be deployed on low-cost hardware.

§ HDFS provides high throughput access to application data and is suitable for applications that have large data sets.

§ HDFS is designed to support very large files. Applications that are compatible with HDFS are those that deal with large data sets. These applications write their data only once but they read it one or more times and require these reads to be satisfied at streaming speeds. HDFS supports write-once-read-many semantics on files.

20. What is HDFS Block size? How is it different from traditional file system block size?

In HDFS data is split into blocks and distributed across multiple nodes in the cluster. Each block is typically 64Mb or 128Mb in size. Each block is replicated multiple times. Default is to replicate each block three times. Replicas are stored on different nodes. HDFS utilizes the local file system to store each HDFS block as a separate file. HDFS Block size cannot be compared with the traditional file system block size.

21. What is a NameNode? How many instances of NameNode run on a Hadoop Cluster?

The NameNode is the centerpiece of an HDFS file system. It keeps the directory tree of all files in the file system, and tracks where across the cluster the file data is kept. It does not store the data of these files itself. There is only One NameNode process run on any hadoop cluster. NameNode runs on its own JVM process. In a typical production cluster its run on a separate machine. The NameNode is a Single Point of Failure for the HDFS Cluster. When the NameNode goes down, the file system goes offline. Client applications talk to the NameNode whenever they wish to locate a file, or when they want to add/copy/move/delete a file. The NameNode responds the successful requests by returning a list of relevant DataNode servers where the data lives.

22. What is a DataNode? How many instances of DataNode run on a Hadoop Cluster? A DataNode stores data in the Hadoop File System HDFS. There is only One DataNode process run on any hadoop slave node. DataNode runs on its own JVM process. On startup, a DataNode connects to the NameNode. DataNode instances can talk to each other. This is mostly during replicating data.

23. How the Client communicates with HDFS?

The Client communication to HDFS happens using Hadoop HDFS API. Client applications talk to the NameNode whenever they wish to locate a file, or when they want to add/copy/move/delete a file on HDFS. The NameNode responds the successful requests by returning a list of relevant DataNode servers where the data lives. Client applications can talk directly to a DataNode, once the NameNode has provided the location of the data.

24. How the HDFS Blocks are replicated?

HDFS is designed to reliably store very large files across machines in a large cluster. It stores each file as a sequence of blocks; all blocks in a file except the last block are the same size. The blocks of a file are replicated for fault tolerance. The block size and replication factor are configurable per file. An application can specify the number of replicas of a file. The replication factor can be specified at file creation time and can be changed later. Files in HDFS are write-once and have strictly one writer at any time. The NameNode makes all decisions regarding replication of blocks. HDFS uses rack-aware replica placement policy. In default configurations there are total 3 copies of a datablock on HDFS, 2 copies are stored on datanodes on same rack and 3rd copy on a different rack.

These are some other sample questions (don’t have answers) from Cloudera certification site.

25. You use the hadoop fs -put command to add sales.txt to HDFS. This file is small enough that it fits into a single block, which is replicated to three nodes within your cluster. When and how will the cluster handle replication following the failure of one of these nodes?

A. The cluster will make no attempt to re-replicate this block.

B. This block will be immediately re-replicated and all other HDFS operations on the cluster will halt while this is in progress.

C. The block will remain under-replicated until the administrator manually deletes and recreates the file.

D. The file will be re-replicated automatically after the NameNode determines it is under-replicated based on the block reports it receives from the DataNodes.

26. You need to write code to perform a complex calculation that takes several steps. You have decided to chain these jobs together and develop a custom composite class for the key that stores the results of intermediate calculations. Which interface must this key implement?

A. Writable

B. Transferable

C. CompositeSortable

D. WritableComparable

27. You are developing an application that uses a year for the key. Which Hadoop-supplied data type would be most appropriate for a key that represents a year?

A. Text

B. IntWritable

C. NullWritable

D. BytesWritable

E. None of these would be appropriate. You would need to implement a custom key.

Hadoop setup single user on centOS

1. Download and install VMware player

Download VMware player latest version and install on your laptop/desktop. You can also install VMware tools, which will be very useful for your working with guest OS.

1.1 Install VMware Tools

You can install VMware tools so that you can you can share data from host system to guest system. You can see full system view only if you install VMware tools.

# make a mount point if needed:

sudo mkdir /media/cdrom

# Mount the CD

sudo mount /dev/cdrom /media/cdrom

#Download the VMWare tools from VMWare player.

Go to Player -> manage - > Install VMWare tools

# Copy and extract VMWareTools

sudo cp /media/cdrom/VMwareTools*.tar.gz ~/Desktop

# you can extract with archive manager, right click on the archive and extract ... or

sudo tar -xvf VMwareTools*.tar.gz

# Install as below

Open a terminal window, and run the following commands.

cd ~/Desktop/vmware-tools-distrib

sudo ./vmware-install.pl

#Enable sharing between the host OS and guest OS as follows:

Go to Virtual Machine – Virtual Machine Settings – Options – Shared Folders and see if the share is enabled or not. If sharing is enabled, you can add a new share location and enjoy working on Hadoop.

2. Download CentOS iso DVD version 6.3, which is stable as on Feb 2013

http://mirrors.hns.net.in/centos/6.3/isos/x86_64/.

If you are using 32bit windows OS then you may download 32bit centos from http://wiki.centos.org/Download

3. Run the VMware player and click on ‘Create New Virtual Machine.

Browse to the iso downloaded in previous step.

CentOS will be installed on local system. We will be using the root user only for installing and running Hadoop.

4. Install Java

Hadoop needs to have Java installed on your CentOS but CentOS does not come with Oracle Java because of licensing issues. So, please use the below commands to install java.

Download the latest java from http://www.oracle.com/technetwork/java/javase/downloads/index.html

And it gets downloaded to Download folder of root user run the following commands.

a) rpm -Uvh /root/Downloads/jdk-7u13-linux-x64.rpmsudo apt-get update

b) alternatives --install /usr/bin/java java /usr/java/latest/jre/bin/java 20000

c) export JAVA_HOME="/usr/java/latest"

d) Confirm the java path by running

javac –version

java -version

5. Confirm your machine name

When you create a new CentOS machine, the default host name is ‘localhost.localdomain.

Open the /etc/hostname file with vi editor and change it if needed.

6. Configuring SSH

Hadoop requires SSH access to manage its nodes, i.e. remote machines plus your local machine if you want to use Hadoop on it (which is what we want to do in this short tutorial). For our single-node setup of Hadoop, we therefore need to configure SSH access to localhost for the anil user we created in the previous section.

#Install ssh server on your computer

yum install openssh-server

ssh-keygen -t rsa -P ""

Generating public/private rsa key pair.

Enter file in which to save the key (/home/anil/.ssh/id_rsa):

Created directory '/home/anil/.ssh'.

Your identification has been saved in /home/anil/.ssh/id_rsa.

Your public key has been saved in /home/anil/.ssh/id_rsa.pub.

The key fingerprint is:

9b:82:ea:58:b4:e0:35:d7:ff:19:66:a6:ef:ae:0e:d2 root@localhost

The key's randomart image is:

[...snipp...]

The final step is to test the SSH setup by connecting to your local machine with the anil user. The step is also needed to save your local machines host key fingerprint to the anil user’s known_hosts file. If you have any special SSH configuration for your local machine like a non-standard SSH port, you can define host-specific SSH options in $HOME/.ssh/config (see man ssh_config for more information).

#now copy the public key to the authorized_keys file, so that ssh should not require passwords every time

root@localhost:~$cat ~/.ssh/id_rsa.pub >> ~/.ssh/authorized_keys

If ssh is not running, then run it by giving the below command

root@localhost:~$service sshd start

root@localhost:~$ ssh localhost

The authenticity of host 'localhost (::1)' can't be established.

RSA key fingerprint is d7:87:25:47:ae:02:00:eb:1d:75:4f:bb:44:f9:36:26.

Are you sure you want to continue connecting (yes/no)? yes

Warning: Permanently added 'localhost' (RSA) to the list of known hosts.

Linux ubuntu 2.6.32-22-generic #33-Ubuntu SMP Wed Apr 28 13:27:30 UTC 2010 i686 GNU/Linux

Ubuntu 10.04 LTS

[...snipp...]

root@localhost:~$

You should get connected to localhost and please exit from ssh by typing exit at command promt.

Next sections will describe how to setup a single-node Hadoop cluster.

7. Download Hadoop

For this tutorial, I am using Hadoop version hadoop-1.0.4, but it should work with any other version.

Got to the URL http://archive.apache.org/dist/hadoop/core/ and click on  hadoop-1.0.4/ and then select  hadoop-1.0.4.tar.gz. The file will be saved to /root/Downloads if you choose defaults. Now perform the following steps to install Hadoop on your CentOS.

Copy your downloaded file from Downloads folder to /usr/local folder

$ cd /usr/local

$ gunzip  hadoop-1.0.4.tar.gz

$ tar -xvf  hadoop-1.0.4.tar

$ ln –s hadoop-1.0.4 hadoop

8. Add Java location to Hadoop so that it can recognize Java

Add the following to /usr/local/hadoop/conf/hadoop-env.sh

export HADOOP_OPTS=-Djava.net.preferIPv4Stack=true

export HADOOP_HOME_WARN_SUPPRESS="TRUE"

export JAVA_HOME=/usr/java/default

9. Update $HOME/.bashrc

Add the following lines to the end of the $HOME/.bashrc file of user anil. If you use a shell other than bash, you should of course update its appropriate configuration files instead of .bashrc.

# Set Hadoop-related environment variables

export HADOOP_HOME=/usr/local/hadoop

# Set JAVA_HOME (we will also configure JAVA_HOME directly for Hadoop later on)

export JAVA_HOME=/usr/java/default

# Some convenient aliases and functions for running Hadoop-related commands

unalias fs &> /dev/null

alias fs="hadoop fs"

unalias hls &> /dev/null

alias hls="fs -ls"

# If you have LZO compression enabled in your Hadoop cluster and

# compress job outputs with LZOP (not covered in this tutorial):

# Conveniently inspect an LZOP compressed file from the command

# line; run via:

# $ lzohead /hdfs/path/to/lzop/compressed/file.lzo

# Requires installed 'lzop' command.

lzohead () {

    hadoop fs -cat $1 | lzop -dc | head -1000 | less

# Add Hadoop bin/ directory to PATH

export PATH=$PATH:$HADOOP_HOME/bin:$PATH:$JAVA_HOME/bin

Close this terminal and open a new one.

10. Create a temporary directory which will be used as base location for DFS.

Now we create the directory and set the required ownerships and permissions:

$ mkdir -p /app/hadoop/tmp

If you forget to set the required ownerships and permissions, you will see a java.io.IOException when you try to format the name node in the next section).

11. Update core-site.xml file

Add the following snippets between the <configuration> ... </configuration> tags in /usr/local/hadoop/conf/core-site.xml:

<!-- In: conf/core-site.xml -->

<property>

  <name>hadoop.tmp.dir</name>

  <value>/app/hadoop/tmp</value>

  <description>A base for other temporary directories.</description>

</property>

<property>

  <name>fs.default.name</name>

  <value>hdfs://localhost:54310</value>

  <description>The name of the default file system.  A URI whose

  scheme and authority determine the FileSystem implementation.  The

  uri's scheme determines the config property (fs.SCHEME.impl) naming

  the FileSystem implementation class.  The uri's authority is used to

  determine the host, port, etc. for a filesystem.</description>

</property>

12. Update mapred-site.xml file

Add the following to /usr/local/hadoop/conf/mapred-site.xml between <configuration> ... </configuration>

<property>

  <name>mapred.job.tracker</name>

  <value>localhost:54311</value>

  <description>The host and port that the MapReduce job tracker runs

  at.  If "local", then jobs are run in-process as a single map

  and reduce task.

  </description>

</property>

13. Update hdfs-site.xml file

Add the following to /usr/local/hadoop/conf/hdfs-site.xml  between <configuration> ... </configuration>

<property>

  <name>dfs.replication</name>

  <value>1</value>

  <description>Default block replication.

  The actual number of replications can be specified when the file is created.

  The default is used if replication is not specified in create time.

  </description>

</property>

14. Format your namenode

Format name with

root@localhost:~$ hadoop namenode –format

15. Starting your single-node cluster

Congratulations, your Hadoop single node cluster is ready to use. Test your cluster by running the following commands.

root@localhost:~$ start-dfs.sh

root@localhost:~$ start-mapred.sh

Check if the Hadoop services are running by opening a browser and hit the below URLs

http://localhost:50030

- for mapreduce

http://localhost:50070

- For hdfs

A command line utility to check whether all the hadoop demons are running or not is: jps

Give the jps at command prompt and you should see this.

root@localhost:~$ jps

9168 Jps

9127 TaskTracker

8824 DataNode

8714 NameNode

8935 SecondaryNameNode

9017 JobTracker