Using HBase

HBase is a NoSQL database from the hadoop family. The NoSql concept is discussed in my blog at What is NoSql ? HBase is a column oriented key value store based on Google's Bigtable.

To recap,  you would be considering a NoSql database because your RDBMS is probably not able to meet your requirements because of one or more of the following reasons:

• You application deals with billions and billion of rows of data

• Application does a lot of writes

• Reads require low latency

• linear scalability with commodity hardware is required

• You frequently need to add more columns or remove columns

There are several NoSql databases that can address one or more of these issues. In this article I provide an introduction to HBase. The goal is to help you get started evaluating whether HBase would be appropriate for your problem. This is introductory material. More details in subsequent blogs.

Main features of HBase are :

• Built on hadoop and HDFS. If you are already using hadoop , then HBase can be viewed as an extension to your hadoop infrastructure that provides random reads and writes.

•  A Simple data model based on keys , values and columns. More on this later. 

• Scales linearly by adding commodity hardware 

• Automatic partitioning of tables as they grow larger 

• Classes available for integration with MapReduce 

• Automatic failover support 

• Support rowkey range scans 

Data Model
 
The main constructs of the model are  Table, rows, column family and columns.

Data is written and read from a Table. A Table has rows and column families. Each row has a key.

Each Column family has one or more columns. Columns in a column family are logically related. Each column has a name and value. When a Table is created, the column families have to be declared. But the columns in each family do not need to be defined and can be added on demand. Each column is referred to using the syntax columnFamily:column. For example, an age column in a userprofile column family is referred to as userprofile:age. For each row, storage space is taken up only for the columns written in that row.

Let us design a Hbase table to store User web browsing information. Each user has a unique id called userid. For each user we need to store

(1) some profile information like sex, age, geolocation, membership.
(2) For each partner website he visits, store the page types viewed, products viewed.
(3) For each partner website he visits, store products purchased , product put in shopping cart but not purchased.

Our structure might look like

{
userid1:{ // rowkey
    profile:{ // column family
          sex: male, // column , value
          age : 25,

          member: Y 

    },
    browsehistory: { // column family
          partner1.hp:23,    // visited partner1 homepage 23 times
          partner2.product.pr1 : 4 // viewed product pr1 4 times
    }

    shoppinghistory: { // column family

         partner3.pr3: 25.5 , // purchased pr3 from partner3 for $25.5

    } 

 }

 Let us design an Hbase table for the above structure.

Tablename : UserShoppingData. Since we will lookup data based on user, the key can be userid.

(1) ColumnFamily profile for profile information. Columns would be sex, age, member etc
(2) ColumnFamily browsehistory for browsing data. Columns are dynamic such as websitename.page or website.productid
(3) ColumnFamily shopping history for shopping data. Columns are dynamic.


The beauty is you can dynamically add columns. If visualizing this as columns is difficult, just think that you are dynamically adding key value pairs.  This kind of data is required in a typical internet shopper analytics application. 

HBase is an appropriate choice because you have several hundred million internet shoppers. That is several million rows. If you wanted to store data by date, you might make the key userid+date, in which case you might have even more rows - in the order of billions. Data is written as the user visits various internet shopping websites. Later the data might need to read with low latency to be able to show the user a promotion or advertisement based on his past history. A company I worked for in the past used a very popular RDBMS for such high volume writes and when ever the RDBMS was flooded with such write requests, the RDBMS would grind to a halt.

Let us use HBase shell to create the above table, insert some data into it and query it. 

Step 1: Download and install HBase from http://hbase.apache.org

Step 2: Start hbase
$ ./start-hbase.sh
starting master, logging to /Users/jk/hbase-0.94.5/bin/../logs/hbase-jk-master-jk.local.out  

Step 3: Start hbase shell
$ ./hbase shell
HBase Shell; enter 'help' for list of supported commands.
Type "exit" to leave the HBase Shell
Version 0.94.5, r1443843, Fri Feb  8 05:51:25 UTC 2013
hbase(main):001:0>
 
Step4: Create the table
hbase(main):004:0> create 'usershoppingdata','profile','browsehistory','shophistory'
0 row(s) in 3.9940 seconds

Step5: Insert some data
hbase(main):003:0> put 'usershoppingdata', 'userid1','profile:sex','male'
0 row(s) in 0.1990 seconds

hbase(main):004:0> put 'usershoppingdata', 'userid1','profile:age','25'
0 row(s) in 0.0090 seconds

hbase(main):005:0> put 'usershoppingdata', 'userid1','browsehistory:amazon.hp','11'
0 row(s) in 0.0100 seconds

hbase(main):006:0> put 'usershoppingdata', 'userid1','browsehistory:amazon.isbn123456','3'
0 row(s) in 0.0070 seconds

hbase(main):007:0> put 'usershoppingdata', 'userid1','shophistory:amazon.isbn123456','19.99'
0 row(s) in 0.0140 seconds
 
Step 6: Read the data
hbase(main):008:0> scan 'usershoppingdata'
ROW                        COLUMN+CELL                                                                
 userid1                   column=browsehistory:amazon.hp, timestamp=1362784343421, value=11          
 userid1                   column=browsehistory:amazon.isbn123456, timestamp=1362786676092, value=3   
 userid1                   column=profile:age, timestamp=1362784243334, value=25                      
 userid1                   column=profile:sex, timestamp=1362784225141, value=male                    
 userid1                   column=shophistory:amazon.isbn123456, timestamp=1362786706557, value=19.99 
1 row(s) in 0.1450 seconds
 
hbase(main):010:0> get 'usershoppingdata', 'userid1'
COLUMN                     CELL                                                                       
 browsehistory:amazon.hp   timestamp=1362784343421, value=11                                          
 browsehistory:amazon.isbn timestamp=1362786676092, value=3                                           
 123456                                                                                               
 profile:age               timestamp=1362784243334, value=25                                          
 profile:sex               timestamp=1362784225141, value=male                                        
 shophistory:amazon.isbn12 timestamp=1362786706557, value=19.99                                       
 3456                                                                                                 
5 row(s) in 0.0520 seconds
 
hbase(main):011:0> get 'usershoppingdata', 'userid1', 'browsehistory:amazon.hp'
COLUMN                     CELL                                                                       
 browsehistory:amazon.hp   timestamp=1362784343421, value=11                                          
1 row(s) in 0.0360 seconds

Step 7: Add few more rows

hbase(main):015:0> put 'usershoppingdata', 'userid2','profile:sex','male'
0 row(s) in 0.0070 seconds

hbase(main):016:0> put 'usershoppingdata', 'userid3','profile:sex','male'
0 row(s) in 0.0060 seconds

hbase(main):017:0> put 'usershoppingdata', 'userid4','profile:sex','male'
0 row(s) in 0.0330 seconds

hbase(main):018:0> put 'usershoppingdata', 'userid5','profile:sex','male'
0 row(s) in 0.0050 seconds

Step 8: Let us do some range scans on the row key
hbase(main):024:0> scan 'usershoppingdata', {STARTROW => 'u'}
ROW                        COLUMN+CELL                                                                
 userid1                   column=browsehistory:amazon.hp, timestamp=1362784343421, value=11          
 userid1                   column=browsehistory:amazon.isbn123456, timestamp=1362786676092, value=3   
 userid1                   column=profile:age, timestamp=1362784243334, value=25                      
 userid1                   column=profile:sex, timestamp=1362784225141, value=male                    
 userid1                   column=shophistory:amazon.isbn123456, timestamp=1362786706557, value=19.99 
 userid2                   column=profile:sex, timestamp=1362788377896, value=male                    
 userid3                   column=profile:sex, timestamp=1362788385501, value=male                    
 userid4                   column=profile:sex, timestamp=1362788392575, value=male                    
 userid5                   column=profile:sex, timestamp=1362788398087, value=male                    
5 row(s) in 0.0780 seconds

hbase(main):019:0> scan 'usershoppingdata', {STARTROW => 'userid3'}
ROW                        COLUMN+CELL                                                                
 userid3                   column=profile:sex, timestamp=1362788385501, value=male                    
 userid4                   column=profile:sex, timestamp=1362788392575, value=male                    
 userid5                   column=profile:sex, timestamp=1362788398087, value=male                    
3 row(s) in 0.0250 seconds

hbase(main):023:0> scan 'usershoppingdata', {STARTROW => 'userid3', STOPROW => 'userid5'}
ROW                        COLUMN+CELL                                                                
 userid3                   column=profile:sex, timestamp=1362788385501, value=male                    
 userid4                   column=profile:sex, timestamp=1362788392575, value=male                    
2 row(s) in 0.0160 seconds

The shell is very useful to playaround with the data model and get familiar with HBase. In a real world application , you might write code in a language like Java. There is more to HBase than this simple introduction. I will get into internals and architecture in future blogs.

Hadoop Secondary Sort: Sorting values

Sorting is a core strength of the Hadoop MapReduce framework. But by default it sorts only the keys. The values for each key are not sorted. There are many use cases where the values for a key are required in a sorted order. For example, let us say a web application logs users interaction - user id, time and other log data. The log data is distributed across log files from different servers. The requirement is to get users log records in  a chronological order.

The input to the map is the set of log records. Let us say (userid,time) not in any order.

The output from the reducer needs to be sorted by user by time.

user1, t1
user1, t2
user1, tn
.
.
usern, t1
usern,tn

where for each user, t1 < t2 ..... < tn.

We know Hadoop sorts the keys. We could make the map output a key that is a combination of user and time. In other words, a composite key that is a combination of userid and time. We need to write a comparator that hadoop can use to sort using the composite key. However a side effect of this is that hadoop will send records for the same user to different reducers. This means you will be not be able to reduce all the users records as a group.

Remember, the hadoop framework sorts the output of a Map by key. It then partitions the output. Each partition is intended for a Reducer.  All values for a key from a Map are in the same partition. We want all the records for a user to go to the same Reducer. This implies they need to be in the same partition. Fortunately there is a way to influence partitioning. We tell hadoop to put all records for a user in the same partition by implementing a partioner that partitions just based on user id and not the composite key.

A Reducer receives partitions from several Mappers. Remember that reducer is called with a key and list of values for that key. Before the framework can call the reducer, it has to group the values for that key from all the partitions.  We want the grouping to happen based on userid ( not userid + time).
So we need to implement a grouping comparator that hadoop uses for grouping and this will compare userids. Since the records in each partition are sorted by userid and time, grouping which is a merge process preserves the sort order - like a merge sort.

In summary you need to

Step 1: Make the map output key a composite of the natural key and value
 Make the Map ouput key a composite of the natural key (userid) and the value (time). A composite key implements a WritableComparable. You need to override the compareTo method to use both userid and time. This method is used to order the keys using the composite key. You also need to override the write and readFields method which are called for serialization and deserialization.
You tell Hadoop to use this composite key by calling the method
job.setOutputKeyClass(UserTime.class) ;

Map output would be like

(user1,t1) , t1
(user1,t2),t2
.
(user2,t1),t1
(user2,t2),t2
 
Step 2: Partition Map output using only the natural key
To ensure all values for the key go the same reducer, you need to implement a partitioner. This is a class that extends org.apache.hadoop.mapreduce.Partitioner. Override the getPartition method to return a partition based on the natural key user id. You tell hadoop to partition using this partitioner with the call
job.setPartitionerClass(NaturalKeyPartitioner.class) ;

Partition from Map 1:
(user1,t1),t1
(user1,t2),t2
(user2,t1),t1
(user2,t2),t2

Partition from Map 2:
(user1,t3),t3
(user1,t4),t4
(user2,t3),t3
(user2,t4),t4


Step 3: Group values using only the natural key
To ensure the reducer gets called with all the values for the key, you need to implement a WritableComparator based on the natural key user id. Override the compare method to compare based on the userid. You tell hadoop to use this comparator for grouping with the call
job.setGroupingComparatorClass(NaturalKeyGroupComparator.class) ;

Input to Reducer 1:
key = (user1,t1), values = t1,t2,t3,t4

Input to Reducer 2:
key = (user2,t1), values = t1,t2,t3,t4


Output from Reducer1:
user1, t1
user1, t2
user1, t3
user1, t4

The complete sample source code is at SecondarySort.jar.

Generics : Array Creation

How do you write a method to convert a generic collection to an Array ? A naive implementation would be:

 
public static <T> T[] convertToArray(Collection<T> c) {

   T[] a = new T[c.size()] ; // compilation error
   int i = 0 ;
   for (T x : c) {
      a[i++] = x ;

   }
     
} 

The code does not compile because the type of Array is required to be able to create an array. Array is what is known as a reifiable type. A type is reifiable if its type information is available at runtime. Any java class or primitive type is reifiable.
Generics on the other hand are implemented by erasure - that is the type information is erased and runtime uses casts to get appropriate behavior. So while

 
List<T> a = new ArrayList<T> () ; 

works because T is erased. Under the hood, just an ArrayList() is created and casts added when getting T. However,

 
T[] a = new T[size] ; // compile error
 

will not work because for arrays type information is required.
The solution is to use reflection, which is what you would if you wanted to dynamically create an instance of any reifiable type like a plain java class. The method signature in our example changes a little to take the array type as an additional parameter. Since it is easier to pass in the required array

 
public static <T> T[] convertToArray(Collection<T> c, T arry) {
   if (arry.length < c.size()) {

      arry = (T[]) java.lang.reflect.Array.newInstance(
                             arry.getClass().getComponentType(),c.size()) ;

      int i = 0 ;
      for (T x : c) {
          a[i++] = x ;
      }
   }
     
} 

The newInstance method on Array creates an array of the required type. getComponentType returns the type of elements of the array. This is analagous to using reflection to a create an instance of class K. You would do

 
K.getClass.newInstance() ;
 

In summary, in generic methods, you can use new operator to create non-reifiable types (eg List<T>) because the type information is erased during compilation (List is created). But for reifiable type, you need to use reflection because the type information is required and cannot be erased

JAVA Garbage Collection

Garbage (GC) collection is the process by which the java virtual machine frees up memory by releasing the memory taken up by objects that are no longer referenced by any other objects. Garbage collection is automatic. For simple applications, the developer does even need to be aware of garbage collection. But for applications with large memory footprint or are long running or have low latency requirements, some understanding is necessary to ensure that garbage collection does not interfere with the application. A common interference of garbage collection is that the application seems to stop responding or the time to respond goes up randomly. The articles lists a few important points every Java developer needs to know about garbage collection.

1.0  Generational GC

Since JDK 5 , the garbage collectors are what are called generational collectors. The heap is divided into regions based on the age of the objects. The young generation has objects that are short lived. The tenured generation has objects that are long lived. All objects are first created in the young region and after a while if they are alive, they are moved to the tenured generation. Garbage collection of the young region happens frequently and is generally fast. GC for the tenured region happens less frequently. Since most objects are short lived, this makes the GC more efficient.

2.0  Types of collectors

Serial Collector : Garbage from both young and tenured regions is done serially and while this happens your application is paused. This is the default collector on single cpu machines and for small heaps sizes ( less that 2G) . This is fine if your application does not care about pauses.

Parallel Collector: This is the default collector on server class machines ( multiple CPUs and greater than 2G heap size). Multiple threads/cpus are used to do garbage collection in parallel for the young region. This makes collection faster. But the application is still paused when GC happens. For the tenured region, the GC is serial as in a serial collector.

Parallel Compacting Collector: GC for the young region is the same as parallel collector and uses multiple threads. However GC for tenured region happens in parallel using multiple CPUs. Application is paused when GC happens.

Concurrent Mark Sweep Collector (CMS): For young region, it is same as in parallel collectors. But for tenured region,  most of the time, GC runs concurrently with the application. The application pauses during GC are expected to be much shorter than the other collectors. This is an ideal choice for applications that cannot tolerate long pauses.

3.0 Understanding GC in your application

Before you try to tune your applications GC, it is important to understand when GC is happening, how much time it takes and how much memory it is reclaiming. The JVM provides the following options to log GC activity.

The -XX:+PrintGCDetails prints GC details described below. The -XX:+PrintGCTimeStamps prints the time from the start of the JVM to when each GC happened. The -Xloggc:gcfilename.log writes the log to gcfilename.log.

In the gc log, you will see a number of lines like

11.561: [GC [PSYoungGen: 868524K->294158K(1198848K)] 1303221K->728855K(4694144K), 0.3640750 secs] [Times: user=1.44 sys=0.02, real=0.37 secs]

This indicates that a GC of the young region occurred at time 11.561 secs from start. The young region was reduced from 868524k to 294158k (66%).  The number (1198848K) is the memory allocated to the young region. The total heap was reduced from 1303221K to 728855K or 44%. The number (4694144K) is the total heap. This GC took .37 secs.

You will see a few lines like

3602.170: [Full GC (System) [PSYoungGen: 16250K->0K(1662080K)] [PSOldGen: 1594630K->1578665K(3495296K)] 1610881K->1578665K(5157376K) [PSPermGen: 22314K->22314K(35904K)], 3.4836190 secs] [Times: user=3.45 sys=0.03, real=3.48 secs]

This indicates that a full GC occurred at 3602.17 secs from the start. The young region was reduced from 16250K to 0K. The old or tenured region was reduced from  1594630K to 1578665K. The total heap was reduced from 1610881K to 1578665K. The GC took 3.48 sec.

The GCViewer is free tool to view GC logs graphically.

GC log viewed in GCViewer

The very small black lines at the bottom indicate the small GCs. The tall black lines at the hourly mark are the Full GCs. The blue peaks are lines indicating how the used heap goes up and goes down after a GC. The ruby red line just below the blue spikes shows the growth of the tenured region. You can see that the tenured region drops after a full GC. Full GCs take a lot of time and you want to reduce the frequency with which they occur.

4.0 Tuning options

 The JVM offers a few knobs that one can turn to tune the GC in a way most suitable to your machine and your application.

-Xms -Xmx options are used to set the initial and maximum size of the heap.  Maximum heap size should be less that physical memory on the machine to avoid paging and one should also leave aside memory for the operating system and other applications running on the same machine. While bigger heap and more memory are good because the GC has to collect less often, when it does have to collect, it has to do more work and the GC pauses could be longer.

–XX:+UseSerialGC
–XX:+UseParallelGC
–XX:+UseParallelOldGC
–XX:+UseConcMarkSweepGC

These options are used to select the GC. SerialGC and ParallelGC are selected by default depending on machine type as described earlier.  Applications that have low latency requirements and cannot tolerate long GC pauses should consider switching to the Concurrent Mark Sweep GC.

-XX:NewSize=n is used to set the default initial size of the young generation. Most applications have many short lived objects and few long lived objects. The newsize should be large enough that short lived objects fit into the young generation and are garbage collected in the small GCs. If the young generation is too small, short lived object get moved to the tenured region which leads to longer Full GCs.

-XX:MaxPauseTimeMillis is a hint to the GC as to the desired maximum pause time. This is just a hint and may or may not be honoured.

5.0 References

There are many other tuning options and the following documents from Oracle are good references on tuning options as well as garbage collection in general:

1. http://www.oracle.com/technetwork/java/javase/gc-tuning-6-140523.html
2. http://www.oracle.com/technetwork/java/javase/tech/memorymanagement-whitepaper-1-150020.pdf

Spring JAVA config tutorial

The classic way of configuring beans in Spring is using XML. But many programmers find switching between XML and java code annoying. Having to go into XML to debug dependencies and track down implementation classes has turned many programmers away from Spring. Since version 3.0, Spring has supported the ability to do configuration using classes and annotations without the need to use XML. In XML , to define a bean, you added to the application.xml
<bean id="Hello" class="com.mj.Hello"/>
To use the bean you wrote code like

ApplicationContext ac = new ClassPathXmlApplicationContext("application.xml") ; 
BeanFactory bf = (BeanFactory) ac ; 
Hello h = bf.getBean("Hello")
h.someMethod() ;

Let us write a new spring application using no XML.

Step1: Define the bean interface and implementation

public interface Greeting {
    public String getMessage() ;
}

public class NewYearGreeting implements Greeting {
    public String getMessage() {
        return "Happy New Year" ;
    }
}
public class BirthDayGreeting  implements Greeting {
    public String getMessage() {
        return "Happy Birthday" ;
    }
}

Step 2: Define the bean configuration in JAVA
The bean definitions are created by writing a class and annotating it with @Configuration. The individual beans are defined by annotating the method that creates the bean with @Bean.

@Configuration
public class GreetingSpringConfig {
    @Bean(name="newyear")
    public Greeting newyearGreeting() {
        return new NewYearGreeting() ;
    }
    @Bean(name="birthday")
    public Greeting birthdayGreeting() {
        return new BirthDayGreeting() ;
    }
 } 

Step 3: Use the beans from a client

 public class GreetingSample {
    public static void main(String args[]) {
        ApplicationContext ac = new    
        AnnotationConfigApplicationContext(GreetingSpringConfig.class) ;
        Greeting g = (Greeting) ac.getBean("newyear") ;
        System.out.println(g.getMessage()) ; 
        g = (Greeting) ac.getBean("birthday") ;
        System.out.println(g.getMessage()) ; 
} 

Note that instead of using ClassPathXmlApplicationContext ,we used AnnotationConfigApplicationContext. AnnotionConfigApplicationContext can process not just @Configuration annotated classes, but also JSR 330 annotated classes. If you don'nt like switching between JAVA & XML , then Java config is simple way of wiring your spring beans.

JAVA Synchronized HashMap vs ConcurrentHashMap

A synchronized HashMap is a Map returned by calling  synchronizedMap methods of java.util.Collections class.

Map syncMap = Collections.synchronizedMap(new HashMap()) ;

The characteristics of synchronized collections are:

1. Each method is synchronized using an object level lock. So the get and put methods on syncMap acquire a lock on syncMap.

2. Compound operations such as check -then - update or iterating over the collection require the client to explicitly acquire a lock on the collection object.

synchronized(syncMap) {
     Integer val = syncMap.get(key) ;
     if ( val == null) {
          syncMap.put(key,  newvalue) ;

}

Without synchronization, multiple threads calling the code can lead to inconsistent values.

3. Locking the entire collection is a performance overhead. While one thread holds on to the lock, no other thread can use the collection.

4. HashMap and other collections from java.util.collections throw ConcurrentModificationException if a thread tries to modify a collection while another thread is iterating over it. The recommended approach is to acquire a lock before iterating over the map.

ConcurrentHashMap was introduced in JDK 5.

The characteristics of ConcurrentHashMap are:

1. There is no locking at the object level. The locking is at a much finer granularity. For a concurrentHashMap , the locks may be at a hashmap bucket level.

2. The effect of lower level locking is that you can have concurrent readers and writers which is not possible for synchronized collections. This leads to much more scalability.

3. Since there no locking at the object level,  additional atomic methods are provided for some compound operations. The ConcurrentHashMap has methods putIfAbsent, remove, replace all of which require checking a key or value and then performing a put or remove.

The code above can be replaced by

ConcurrentHashMap concMap ;
.
.
.
 concMap.putIfAbsent(key,newvalue) ;

4. ConcurrentHashMap does not throw a ConcurrentModificationException if one thread tries to modify it while another is iterating over it. The iterator returned by ConcurrentHashMap is an iterator on a snapshot of the data when the iterator was created. It may or may not have changes made by other threads after the iterator was created.

In general, using ConcurrentHashMap instead of synchronized Map gives you much better scalability and you do not have to explicitly synchronize on the map object.

Hadoop 2.x Tutorial

Hadoop 2.x release involves many changes to Hadoop and MapReduce. The centralized JobTracker service is replaced with a ResourceManager that manages the resources in the cluster and an ApplicationManager that manages the application lifecycle. These architectural changes enable hadoop to scale to much larger clusters. A new release also has minor changes to scripts,directories and environment variables necessary to get started. This is a getting started tutorial for 2.x. The intended audience is someone who is completely new to hadoop and needs a jumpstart or someone who has played a little bit with a previous version and wants to start using 2.x.  The emphasis on getting hadoop running and not necessarily explaining concepts which is covered in many other blogs.

In this tutorial we will

(1) Setup a hadoop in a single node environment
(2) Create and move files to HDFS
(3) Write and execute a simple MapReduce Application

Step 1: Download Hadoop and install
Download the current 2.x.x release from http://hadoop.apache.org/releases.html.
I downloaded hadoop-2.0.1-alpha.tar.gz.
Untar the file to a directory say ~/hadoop-2.0.1-alpha.

Step 2: Set the following environment variables
hadoop-2.0.1-alpha$ export HADOOP_HOME=~/hadoop-2.0.1-alpha
hadoop-2.0.1-alpha$ export HADOOP_MAPRED_HOME=~/hadoop-2.0.1-alpha
hadoop-2.0.1-alpha$ export HADOOP_COMMON_HOME=~/hadoop-2.0.1-alpha
hadoop-2.0.1-alpha$ export HADOOP_HDFS_HOME=~/hadoop-2.0.1-alpha
hadoop-2.0.1-alpha$ export YARN_HOME=~/hadoop-2.0.1-alpha
hadoop-2.0.1-alpha$ export HADOOP_CONF_DIR=~/hadoop-2.0.1-alpha/etc/hadoop

If these environment variables are not setup correctly, Step 4 might fail.

Step 3: Update the configuration files
 hdfs-site.xml
Add the following configuration to etc/hadoop/hdfs-site.xml. If you do not set dfs.namenode.name.dir and dfs.datanote.data.dir explicitly, hadoop will default to a temp directory that the OS may clean up on restart and you will lose data. This is a common omission for newbees.
 <?xml version="1.0" encoding="UTF-8"?>
<configuration>
<property>
    <name>dfs.replication</name>
    <value>1</value>
  </property>
  <property>
    <name>dfs.namenode.name.dir</name>
    <value>file:/Users/joe/hadoop-hdfs201/data/hdfs/namenode</value>
  </property>
  <property>
    <name>dfs.datanode.data.dir</name>
    <value>file:/Users/joe/hadoop-hdfs201/data/hdfs/datanode</value>
  </property>
</configuration>
core-site.xml
Add the following to etc/hadoop/core-site.xml.
<?xml version="1.0" encoding="UTF-8"?>
<configuration>
<property>
    <name>fs.default.name</name>
    <value>hdfs://localhost:9000</value>
  </property>
</configuration>
yarn-site.xml
Add the following to etc/hadoop/yarn-site.xml.
<?xml version="1.0"?>
<configuration>
  <property>
    <name>yarn.nodemanager.aux-services</name>
    <value>mapreduce.shuffle</value>
  </property>
 <property>
    <name>yarn.nodemanager.aux-services.mapreduce.shuffle.class</name>
    <value>org.apache.hadoop.mapred.ShuffleHandler</value>
  </property>
</configuration>
mapred-site.xml
Add the following to etc/hadoop/mapred-site.xml.
<?xml version="1.0"?>
<configuration>
  <property>
    <name>mapreduce.framework.name</name>
    <value>yarn</value>
  </property>
</configuration> 

Step 4: Start the processes.

Change to the directory where hadoop is installed.
cd ~/hadoop-2.0.1-alpha

If you are running hadoop for the first time, the following command will format HDFS. Do not run this everytime as it formats and thus deletes any existing data

hadoop-2.0.1-alpha$ bin/hadoop namenode -format

Start the namenode.
hadoop-2.0.1-alpha$ sbin/hadoop-daemon.sh start namenode

Start the datanode.
hadoop-2.0.1-alpha$ sbin/hadoop-daemon.sh start datanode

In hadoop 2.x , there is no jobtracker. Instead there is a resourcemanager and a nodemanager.
Start the resourcemanager.
hadoop-2.0.1-alpha$ sbin/yarn-daemon.sh start resourcemanager

Start the nodemanager.
hadoop-2.0.1-alpha$ sbin/yarn-daemon.sh start nodemanager

Start the history server.
hadoop-2.0.1-alpha$ sbin/mr-jobhistory-daemon.sh start historyserver

Type jps. It lists the java processes running. Check that all the processes are started

hadoop-2.0.1-alpha$ jps

1380 DataNode
1558 Jps
1433 ResourceManager
1536 JobHistoryServer
1335 NameNode
1849 NodeManager

Once I ran into a problem where the mapreduce job was being accepted but never executed. In looking at the logs, I found that the NodeManager had not started.  The jps command is a good check to ensure all necessary processes are started.

Step 5: Get familiar with HDFS
The HDFS commands are documented in the older releases of hadoop
http://hadoop.apache.org/docs/r1.0.3/file_system_shell.html

hadoop-2.0.1-alpha$ bin/hadoop fs -ls

will list the home directory. If you are user joe. HDFS creates a /user/joe directory for you. Any files or directories you create will be created here.

hadoop-2.0.1-alpha$ bin/hadoop fs -mkdir /user/joe/input
creates a directory input

In the local filesystem create a file app.log with the data

user01|1|2|3|4|5
user02|1|2|3|4|5
user03|1|2|3|4|5
user01|1|2|3|4|5
user02|1|2|3|4|5
user01|1|2|3|4|5
user03|1|2|3|4|5
user01|1|2|3|4|5
user04|1|2|3|4|5
user01|1|2|3|4|5

let us pretend this is a log file from a web application where for each request we have logged userid and some additional data. We will later use this .as input for a MapReduce program.
You can move it to hdfs using the command

hadoop-2.0.1-alpha$ bin/hadoop fs -moveFromLocal ~/projects/app.log /user/manoj/input/

To print the file just moved to hdfs
hadoop-2.0.1-alpha$ bin/hadoop fs -cat /user/manoj/input/app.log

Step 6: Create a MapReduce program
The MapReduce programming model is explained in the blog What is MapReduce ?. Let us write a simple mapreduce program that uses that app.log we created in step5 as input and outputs the number of times a user visited the site. UserCountMap reads a line and outputs (username,1).  UserCountReducer takes as input (username, list of 1s) and outputs (username, sum).

public class UserCount {
    public static class UserCountMap extends Mapper    
        public void map(LongWritable key, Text Value, Context context)
                        throws IOException, InterruptedException {
            String line = Value.toString() ;
            String tokens[] = line.split("\\|") ;           
            if (tokens.length > 0) {               
                context.write(new Text(tokens[0]),new IntWritable(1)) ;               
            }
        }     
    }
    public static class UserCountReducer extends Reducer  
        public void reduce(Text key, Iterable values, Context context)
                    throws IOException, InterruptedException {
                 int count = 0 ;      
            for (IntWritable value : values) {
                count = count + value.get() ;
            }
            context.write(key, new IntWritable(count)) ;
        }   
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = new Job(conf);
        job.setJarByClass(UserCount.class) ;       
        FileInputFormat.addInputPath(job, new Path(args[0])) ;
        FileOutputFormat.setOutputPath(job, new Path(args[1])) ;       
        job.setMapperClass(UserCountMap.class) ;
        job.setReducerClass(UserCountReducer.class) ;       
        job.setOutputKeyClass(Text.class) ;
        job.setOutputValueClass(IntWritable.class) ;
        System.exit(job.waitForCompletion(true) ? 0 : 1) ;       
    }   
      
}

Compile the program and package into a jar called say usercount.jar.

Step 7: Run the madreduce program

hadoop-2.0.1-alpha$ bin/hadoop jar ~/projects/usercount.jar com.mj.UserCount /user/joe/input /user/joe/output

you should see output some of which is shown below.

12/09/12 17:41:28 INFO mapreduce.Job: The url to track the job: http://joe.local:8088/proxy/application_1347494786422_0003/
12/09/12 17:41:28 INFO mapreduce.Job: Running job: job_1347494786422_0003
12/09/12 17:41:37 INFO mapreduce.Job: Job job_1347494786422_0003 running in uber mode : false
12/09/12 17:41:37 INFO mapreduce.Job:  map 0% reduce 0%
12/09/12 17:41:43 INFO mapreduce.Job:  map 100% reduce 0%
12/09/12 17:41:45 INFO mapreduce.Job:  map 100% reduce 100%
12/09/12 17:41:45 INFO mapreduce.Job: Job job_1347494786422_0003 completed successfully

You can see the status of the job at http:/localhost:8088

hadoop-2.0.1-alpha$ bin/hadoop fs -ls /user/joe/output
212/09/12 17:45:29 WARN util.KerberosName: Kerberos krb5 configuration not found, setting default realm to empty
Found 2 items
-rw-r--r--   1 manoj supergroup          0 2012-09-12 17:41 /user/joe/output/_SUCCESS
-rw-r--r--   1 manoj supergroup         39 2012-09-12 17:41 /user/joe/output/part-r-00000

The file part-r-00000 will have the output which is

user01    5
user02    2
user03    2
user04    1

Hoping these steps help jumpstart you with hadoop and get you going on your way to write more complex Map Reduce jobs to analyze your big data.

Update 2/28/2014: For how to setup hadoop 2.x Cluster , see Hadoop 2.x YARN cluster setup tutorial