Twitter's Storm : A Real-Time Hadoop

Operating on HBase columns

HBase is a column oriented database which stores its contents by column rather than by row. Instead of retrieving a record or row at a time, an entire column is retrieved and thus it becomes very powerful and efficient since data analytics is usually concerned with only one field or column of a record. The access becomes much faster, and much more relevant data can be extracted from the database in a shorter period of time.

As already mentioned in my previous post, apart from the basic read, write and delete operations I have developed another set of functions to perform union, intersection on hbase tables and also use having, between and distinct clauses as we do in SQL.
Since HBase is a column oriented database i.e. retrieves an entire column at a time instead of row it becomes very powerful and efficient since data analytics is usually concerned with only one field or column of a record. And for handling columns such functions play a significant role.

The sample program below illustrate the following operations :

Obtaining all distinct entries of a column from a table
Obtaining all distinct entries of a column from a table with the number of occurance of each.
Implemention of the Having operator.
Implementing the Having operator and extracting the entire satisfying rows
Implementation of the Between operator
Implementation of the Union operator
Implementation of the Intersection operator

Program :

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
import org.apache.hadoop.hbase.filter.FilterList.Operator;
import org.apache.hadoop.hbase.util.Bytes;

public class HBaseTest
{
private static HBaseConfiguration conf;
HBaseTest()
    {
        conf = new HBaseConfiguration();
        conf.addResource(new Path("/path_to_your_hbase/hbase-0.20.6/conf/hbase-site.xml"));
    }

// function to obtain distinct col entries from a table.
   
public Set<String> distinct(String tableName,String colFamilyName, String colName)
   {
    Set<String> set = new HashSet<String>();
    ResultScanner rs=null;
    Result res = null;
    String s = null;
    try
    {
        HTable table = new HTable(conf, tableName);
        Scan scan = new Scan();
        scan.addColumn(Bytes.toBytes(colFamilyName),Bytes.toBytes(colName));
        rs = table.getScanner(scan);
        while((res=rs.next()) != null)
        {
            byte [] obtCol = res.getValue(Bytes.toBytes(colFamilyName+":"+colName));               
            s = Bytes.toString(obtCol);
            set.add(s);
        }
    } catch (IOException e)
    {
        System.out.println("Exception occured in retrieving data");
    }
    finally
    {
        rs.close();
    }
    return set;
   }

// function to return distinct entries with the number of occurance of each.
   
public  HashMap<String, Integer> distinctWithOccurances(String tableName,String colFamilyName, String  colName)
   {
    HashMap<String, Integer> map = new HashMap<String,Integer>();
    ResultScanner rs=null;
    Result res = null;
    try
    {
        HTable table = new HTable(conf, tableName);
        Scan scan = new Scan();
        rs = table.getScanner(scan);
        String s = null;
        while((res=rs.next()) != null)
        {
            int noofOccurance = 0;
            int count=0;
            byte [] obtCol = res.getValue(Bytes.toBytes(colFamilyName+":"+colName));               
            s = Bytes.toString(obtCol);
            Set<String> set =  map.keySet();
            Iterator iterator = set.iterator();
            if(iterator != null)
            {
            while(iterator.hasNext() && count==0)
            {
                String colEntry = (String) iterator.next();
                if(colEntry.equals(s))
                {
                noofOccurance = map.get(colEntry);
                int newNoofOccurance = noofOccurance + 1;
                map.put(s,newNoofOccurance);
                count++;
                }
            }
            }
            if(count == 0)
            {
                map.put(s,1);
            }
        }
    } catch (IOException e)
    {
        System.out.println("Exception occured in retrieving data");
    }
    finally
    {
        rs.close();
    }
    return map;
   }

// function implementing having clause.
   
public ArrayList<HashMap<String, String>> having(String tableName,String colFamilyName, String [] colName,String havingColName,String value)
  {
    ResultScanner rs=null;
    ArrayList<HashMap<String, String>> al = new ArrayList<HashMap<String, String>>();
    Result res = null;
    try
    {
        HTable table = new HTable(conf, tableName);
        Scan scan = new Scan();
        SingleColumnValueFilter singleColumnValueFilterA = new SingleColumnValueFilter(
                Bytes.toBytes(colFamilyName), Bytes.toBytes(havingColName), CompareOp.EQUAL, Bytes.toBytes(value));
              singleColumnValueFilterA.setFilterIfMissing(true); 
                FilterList filter = new FilterList(Operator.MUST_PASS_ALL, Arrays
                           .asList((Filter) singleColumnValueFilterA));
                scan.setFilter(filter); 
        rs = table.getScanner(scan);
        while((res=rs.next()) != null)
        {
            HashMap<String, String> map = new HashMap<String,String>();
            String s = null;
            for(int j=0 ; j < colName.length ; j++)
            {
            byte[] obtainedRow = res.getValue(Bytes.toBytes(colFamilyName),Bytes.toBytes(colName[j]));
            System.out.println(colName[j]);
            s = Bytes.toString(obtainedRow);
            map.put(colName[j],s);
        }           
        al.add(map);
        }
    } catch (IOException e)
        {
            System.out.println("Exception occured in retrieving data");
        }
    finally
    {
        rs.close();
    }
        return al;
   }

// function implementing having clause and extracting the entire rows.
   
public ArrayList<HashMap<String, String>> havingWithEntireRow(String tableName,String colFamilyName[], String [][] colName,String havingColFamilyName,String havingColName,String value)
  {
    ResultScanner rs=null;
    ArrayList<HashMap<String, String>> al = new ArrayList<HashMap<String, String>>();
    Result res = null;
    try
    {
        HTable table = new HTable(conf, tableName);
        Scan scan = new Scan();
        SingleColumnValueFilter singleColumnValueFilterA = new SingleColumnValueFilter(
                    Bytes.toBytes(havingColFamilyName), Bytes.toBytes(havingColName), CompareOp.EQUAL, Bytes.toBytes(value));
            singleColumnValueFilterA.setFilterIfMissing(true); 
            FilterList filter = new FilterList(Operator.MUST_PASS_ALL, Arrays
                           .asList((Filter) singleColumnValueFilterA));
                scan.setFilter(filter); 
            rs = table.getScanner(scan);
        while((res=rs.next()) != null)
        {
            HashMap<String, String> map = new HashMap<String,String>();
            String s = null;
            for(int i=0 ; i< colFamilyName.length ; i++)
            {
            for(int j=0 ; j < colName[i].length ; j++)
            {
            byte[] obtainedRow = res.getValue(Bytes.toBytes(colFamilyName[i]),Bytes.toBytes(colName[i][j]));
            s = Bytes.toString(obtainedRow);
            map.put(colName[i][j],s);
            }   
            }
            al.add(map);
        }
    } catch (IOException e)
        {
            System.out.println("Exception occured in retrieving data");
        }
    finally
    {
        rs.close();
    }
        return al;
    }

// function implementing the between clause.
   
public ArrayList<HashMap<String, String>> between(String tableName,String colFamilyName, String [] colName,String betweenColName,String lowerValue,String upperValue)
   {
    ResultScanner rs=null;
    ArrayList<HashMap<String, String>> al = new ArrayList<HashMap<String, String>>();
    Result res = null;
    try
    {
        HTable table = new HTable(conf, tableName);
        Scan scan = new Scan();
        SingleColumnValueFilter singleColumnValueFilterA = new SingleColumnValueFilter(
                Bytes.toBytes(colFamilyName), Bytes.toBytes(betweenColName), CompareOp.GREATER, Bytes.toBytes(lowerValue));
                singleColumnValueFilterA.setFilterIfMissing(true); 
        SingleColumnValueFilter singleColumnValueFilterB = new SingleColumnValueFilter(
                     Bytes.toBytes(colFamilyName), Bytes.toBytes(betweenColName), CompareOp.LESS_OR_EQUAL, Bytes.toBytes(upperValue));
        singleColumnValueFilterB.setFilterIfMissing(true); 
        FilterList filter = new FilterList(Operator.MUST_PASS_ALL, Arrays.asList((Filter) singleColumnValueFilterA,
                                    singleColumnValueFilterB)); 
                scan.setFilter(filter); 
        rs = table.getScanner(scan);
        while((res=rs.next()) != null)
        {
            HashMap<String, String> map = new HashMap<String,String>();
            String s = null;
            for(int j=0 ; j < colName.length ; j++)
            {
            byte[] obtainedRow = res.getValue(Bytes.toBytes(colFamilyName),Bytes.toBytes(colName[j]));
            s = Bytes.toString(obtainedRow);
            map.put(colName[j],s);
            }           
            al.add(map);
        }
    } catch (IOException e)
        {
            System.out.println("Exception occured in retrieving data");
        }
    finally
    {
        rs.close();
    }
            return al;
   }

// function implementing union.

public ArrayList<HashMap<String, String>> union(String tableName,String colFamilyName1, String colFamilyName2,String [] colNames1,String [] colNames2, String colName1, String colName2,String value1,String value2)
   {
    ResultScanner rs=null;
    ArrayList<HashMap<String, String>> al = new ArrayList<HashMap<String, String>>();
    Result res = null;
    try
    {
        HTable table = new HTable(conf, tableName);
        Scan scan = new Scan();
        SingleColumnValueFilter singleColumnValueFilterA = new SingleColumnValueFilter(
                Bytes.toBytes(colFamilyName1), Bytes.toBytes(colName1), CompareOp.EQUAL, Bytes.toBytes(value1));
                singleColumnValueFilterA.setFilterIfMissing(true); 
                 
            SingleColumnValueFilter singleColumnValueFilterB = new SingleColumnValueFilter(
                     Bytes.toBytes(colFamilyName2), Bytes.toBytes(colName2), CompareOp.EQUAL, Bytes.toBytes(value2));
        singleColumnValueFilterB.setFilterIfMissing(true); 
                 
        FilterList filter = new FilterList(Operator.MUST_PASS_ONE, Arrays.asList((Filter) singleColumnValueFilterA,
                                    singleColumnValueFilterB)); 
                     
               scan.setFilter(filter); 
            rs = table.getScanner(scan);
        if(colFamilyName1.equals(colFamilyName2))
        {
        while((res=rs.next()) != null)
        {
            HashMap<String, String> map = new HashMap<String,String>();
            String s = null;
            for(int j=0 ; j < colNames1.length ; j++)
            {
            byte[] obtainedRow = res.getValue(Bytes.toBytes(colFamilyName1),Bytes.toBytes(colNames1[j]));
            System.out.println(colNames1[j]);
            s = Bytes.toString(obtainedRow);
            System.out.println(s);
            map.put(colNames1[j],s);
            }           
            al.add(map);
        }
        }
        else
        {
            while((res=rs.next()) != null)
            {
                HashMap<String, String> map = new HashMap<String,String>();
                String s = null;
                // extract row of the first col family
                for(int j=0 ; j < colNames1.length ; j++)
                {
                byte[] obtainedRow = res.getValue(Bytes.toBytes(colFamilyName1),Bytes.toBytes(colNames1[j]));
                s = Bytes.toString(obtainedRow);
                map.put(colNames1[j],s);
                }           
                // extract row of the second col family
                for(int k=0 ; k < colNames2.length ; k++)
                {
                byte[] obtainedRow = res.getValue(Bytes.toBytes(colFamilyName2),Bytes.toBytes(colNames2[k]));
                s = Bytes.toString(obtainedRow);
                map.put(colNames2[k],s);
                }       
                // put both in the arraylist
                al.add(map);
            }   
        }
    } catch (IOException e)
        {
            System.out.println("Exception occured in retrieving data");
        }
    finally
    {
        rs.close();
    }
    return al;
   }

// function implementing intersection.

public ArrayList<HashMap<String, String>> intersection(String tableName,String colFamilyName1, String colFamilyName2,String [] colNames1,String [] colNames2, String colName1, String colName2,String value1,String value2)
  {
    ResultScanner rs=null;
    ArrayList<HashMap<String, String>> al = new ArrayList<HashMap<String, String>>();
    Result res = null;
    try
    {
        HTable table = new HTable(conf, tableName);
        Scan scan = new Scan();
        SingleColumnValueFilter singleColumnValueFilterA = new SingleColumnValueFilter(
                    Bytes.toBytes(colFamilyName1), Bytes.toBytes(colName1), CompareOp.EQUAL, Bytes.toBytes(value1));
            singleColumnValueFilterA.setFilterIfMissing(true); 
        SingleColumnValueFilter singleColumnValueFilterB = new SingleColumnValueFilter(
                 Bytes.toBytes(colFamilyName2), Bytes.toBytes(colName2), CompareOp.EQUAL, Bytes.toBytes(value2));
                singleColumnValueFilterB.setFilterIfMissing(true); 
                 
        FilterList filter = new FilterList(Operator.MUST_PASS_ALL, Arrays.asList((Filter) singleColumnValueFilterA,
                                    singleColumnValueFilterB)); 
                scan.setFilter(filter); 
        rs = table.getScanner(scan);
        if(colFamilyName1.equals(colFamilyName2))
        {
        while((res=rs.next()) != null)
        {
            HashMap<String, String> map = new HashMap<String,String>();
            String s = null;
            for(int j=0 ; j < colNames1.length ; j++)
            {
            byte[] obtainedRow = res.getValue(Bytes.toBytes(colFamilyName1),Bytes.toBytes(colNames1[j]));
            s = Bytes.toString(obtainedRow);
            map.put(colNames1[j],s);
            }           
            al.add(map);
        }
        }
        else
        {
            while((res=rs.next()) != null)
            {
                HashMap<String, String> map = new HashMap<String,String>();
                String s = null;
                // extract row of the first col family
                for(int j=0 ; j < colNames1.length ; j++)
                {
                byte[] obtainedRow = res.getValue(Bytes.toBytes(colFamilyName1),Bytes.toBytes(colNames1[j]));
                s = Bytes.toString(obtainedRow);
                //System.out.println(s);
                map.put(colNames1[j],s);
                }           
                // extract row of the second col family
                for(int k=0 ; k < colNames2.length ; k++)
                {
                byte[] obtainedRow = res.getValue(Bytes.toBytes(colFamilyName2),Bytes.toBytes(colNames2[k]));
                s = Bytes.toString(obtainedRow);
                map.put(colNames2[k],s);
                }       
                // put both in the arraylist
                al.add(map);
            }   
        }
        } catch (IOException e)
        {
            System.out.println("Exception occured in retrieving data");
        }
        finally
        {   
            System.out.println("in intersection");
            rs.close();
        }
        return al;
   }

public static void main(String args[])
    {
        HBaseTest test  = new HBaseTest();
        String tableName = "testing_table" ;
        String [] colFamilyNames = {"colFamily1","colFamily2"};
        String [][] colNames  = {{"Id","Name"},{"Addr","Designation"}};

    Set<String> set =  test.distinct(tableName,"colFamily1","Name");
    Iterator iterator = set.iterator();
    while(iterator.hasNext())
        {
            String valofKey = (String) iterator.next();
            System.out.println(valofKey + "=" + newMap.get(valofKey));
        }
          
       HashMap<String, Integer> map = new HashMap<String,Integer>();
       map = operationObj.distinctWithOccurances(tableName, "Name", "Designation");

       ArrayList<HashMap<String, String>> al_having = new ArrayList<HashMap<String, String>>();
       al_having = operationObj.havingWithEntireRow(tableName, colFamilyNames, colNames, "colFamily1", "Name", "Jayati");

       String [] reqdFieldNames1 = {"Id","Name"};
       String [] reqdFieldNames2 = {"Id","Name"};
       ArrayList<HashMap<String, String>> al_intersection = new ArrayList<HashMap<String, String>>();
       al_intersection = operationObj.intersection(tableName, colFamily1, colFamily2,reqdFieldNames1,reqdFieldNames2,"Id","Name","1","Jayati");
   
    // similarly union and between can be used
    }
}

Utilize basic read-write functions

HBase is an open source, non-relational, distributed database providing BigTable like capabilities for Hadoop. Tables in HBase can be accessed using the Java-API for HBase but unfortunately a developer would require to put in a lot of efforts to do so. That is because the API provides a very restricted set of functions. For those new to API , it takes a lot of time to understand the available classes and use them to get the required job done.

So to enable easy handling of HBase tables, I have developed a wrapper library over the existing API which provides basic methods to create, read , delete records in hbase table and also another set of functions such as distinct, having, between, intersection, union which work for HBase just as we have these working in SQL. A big fraction of our work on tables depends on these functions and their availability makes using the HBase API easy.

This post includes a sample program to illustrate the usage of read and write functions only which specifically includes the following operations :

Adding entry to a single column
Adding records with a single column family and multiple columns
Adding a row with any number of column families and columns
Obtaining a single column entry
Obtaining the entire row
Reading all entries of a particular column of a table
Reading all records of an HBase Table
Deleting a record from an HBase Table

I have used hbase-0.20.6 and hadoop-0.20.1 and you could deploy this program on your eclipse and make test classes to check it.

Program :

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
import org.apache.hadoop.hbase.filter.FilterList.Operator;
import org.apache.hadoop.hbase.util.Bytes;

public class HBaseTest
{
private static HBaseConfiguration conf;
HBaseTest()
    {
        conf = new HBaseConfiguration();
        conf.addResource(new Path("/path_to_your_hbase/hbase-0.20.6/conf/hbase-site.xml"));
    }

// assigns a value to a particular column of a record

public void addAColumnEntry(String tableName, String colFamilyName, String colName, String data)
    {
        try
        {
            HTable table = new HTable(conf, tableName);
            String row = "row" + Math.random();
            byte[] rowKey = Bytes.toBytes(row);
            Put putdata = new Put(rowKey);
            putdata.add(Bytes.toBytes(colFamilyName), Bytes.toBytes(colName),Bytes.toBytes(data));
            table.put(putdata);
        } catch (IOException e)
        {
            System.out.println("Exception occured in adding data");
        }
    }

// write a record to a table having just one column family or write only a portion of a record

public void addRecordWithSingleColumnFamily(String tableName, String colFamilyName, String [] colName,String [] data)
    {
        try
        {
            HTable table = new HTable(conf, tableName);
            String row = "row" + Math.random();
            byte[] rowKey = Bytes.toBytes(row);
            Put putdata = new Put(rowKey);
            if(colName.length == data.length)
            {
            for(int i=0 ; i < colName.length ; i++)
            putdata.add(Bytes.toBytes(colFamilyName), Bytes.toBytes(colName[i]),
                    Bytes.toBytes(data[i]));
            }
            table.put(putdata);
     
        } catch (IOException e)
        {
            System.out.println("Exception occured in adding data");
        }
    }

// add a record with any number of column families

public void addRecord(String tableName, String [] colFamilyName, String [][]  colName,String [][] data)
    {
        try
        {
            HTable table = new HTable(conf, tableName);
            String row = "row" + Math.random();
            byte[] rowKey = Bytes.toBytes(row);
            Put putdata = new Put(rowKey);
            for(int j=0 ; j < colFamilyName.length ; j++)
            {
            if(colName[j].length == data[j].length)
            {
            for(int i=0 ; i < colName[j].length ; i++)
            putdata.add(Bytes.toBytes(colFamilyName[j]), Bytes.toBytes(colName[j][i]),
                    Bytes.toBytes(data[j][i]));
            }
            }
            table.put(putdata);
     
        } catch (IOException e)
        {
            System.out.println("Exception occured in adding data");
        }
    }

// returns entry of a particular column of a record

public String getColEntry(String tableName, String rowName,String colFamilyName, String colName)
    {
        String result = null;
        try
        {
            HTable table = new HTable(conf, tableName);
            byte[] rowKey = Bytes.toBytes(rowName);
            Get getRowData = new Get(rowKey);
            Result res = table.get(getRowData);
            byte[] obtainedRow = res.getValue(Bytes.toBytes(colFamilyName),
                    Bytes.toBytes(colName));
            result = Bytes.toString(obtainedRow);
        } catch (IOException e)
        {
            System.out.println("Exception occured in retrieving data");
        }
        return result;
    }
 
// returns a row  in the form of a string.
 
public String getRow(String tableName, String rowName,String colFamilyName, String [] colName)
    {
            String result = colName[0];
        try
        {
            HTable table = new HTable(conf, tableName);
            byte[] rowKey = Bytes.toBytes(rowName);
            Get getRowData = new Get(rowKey);
            Result res = table.get(getRowData);
            for(int j=0 ; j < colName.length ; j++)
            {
            byte[] obtainedRow = res.getValue(Bytes.toBytes(colFamilyName),Bytes.toBytes(colName[j]));
            System.out.println(colName[j]);
            String s = Bytes.toString(obtainedRow);
            if(j==0)
                result = colName[j] + "=" + s ;
            else
                result = result + "&" + colName[j] + "=" + s;
            System.out.println(s);
            }
         
        } catch (IOException e)
        {
            System.out.println("Exception occured in retrieving data");
        }
        return result;
    }
 
// returns an arraylist of all entries of a column.
 
public ArrayList<String> getCol(String tableName,String colFamilyName, String colName)
    {
        ArrayList<String> al = new ArrayList<String>();
        ResultScanner rs=null;
        Result res = null;
     
        try {
            HTable table = new HTable(conf, tableName);
         
            Scan scan = new Scan();
            scan.addColumn(Bytes.toBytes(colFamilyName),Bytes.toBytes(colName));
            rs = table.getScanner(scan);
            while((res=rs.next()) != null)
            {
                String colEntry = null;
                byte [] obtCol = res.getValue(Bytes.toBytes(colFamilyName+":"+colName));             
                colEntry = Bytes.toString(obtCol);
                al.add(colEntry);
            }
         
        } catch (IOException e)
        {
            System.out.println("Exception occured in retrieving data");
        }
        finally
        {
            rs.close();
        }
        return al;

    }
 
// returns a list of hashmaps, each hashmap containing entries of a single record.

public  ArrayList<HashMap<String, String>> getTable(String tableName,String [] colFamilyName, String [][] colName)
    {
    ResultScanner rs=null;
    ArrayList<HashMap<String, String>> al = new ArrayList<HashMap<String, String>>();
    Result res = null;
    try
    {
        HTable table = new HTable(conf, tableName);
        Scan scan = new Scan();
        rs = table.getScanner(scan);
        while((res=rs.next()) != null)
        {
            HashMap<String, String> map = new HashMap<String,String>();
            String s = null;
            for(int i=0 ; i<colFamilyName.length ; i++)
            {
                for(int j=0 ; j < colName[i].length ; j++)
                    {
                        byte[] obtainedRow = res.getValue(Bytes.toBytes(colFamilyName[i]),Bytes.toBytes(colName[i][j]));
                        s = Bytes.toString(obtainedRow);
                        System.out.println(s);
                        map.put(colName[i][j],s);
                    }     
            }         
            al.add(map);
        }
    } catch (IOException e)
        {
            System.out.println("Exception occured in retrieving data");
        }
    finally
    {
        rs.close();
    }
        return al;
    }

// function to delete a row from the table.

public String deleteTableRow(String tableName, String rowName)
   {
   String result = null;
   try
   {
    HTable table = new HTable(conf, tableName);
    byte[] rowKey = Bytes.toBytes(rowName);
    Delete delRowData = new Delete(rowKey);
    table.delete(delRowData);
   } catch (IOException e)
    {
        System.out.println("Exception occured in retrieving data");
    }
  return result;

  }

public static void main(String args[])
    {
        HBaseTest test  = new HBaseTest();
        String tableName = "testing_table" ;
        String [] colFamilyNames = {"colFamily1","colFamily2"};
        String [][] colNames  = {{"Id","Name"},{"Addr","Designation"}};
         
        test.addAColumnEntry(tableName,"colFamily1","Name","Ram");
        test.addRecordWithSingleColumnFamily(tableName,"colFamily1",{"Id","Name"},{"117","Ram"});
        test.addRecord(tableName,colFamilyNames,colNames,{{"117","Ram"},{"ABC","Manager"}});

        // specify the rowKey as per your table

        test.getColEntry(tableName,rowKey,"colFamily1","Name");
        String yourRow = test.getRow(tableName,"row0.35234564623454","colFamily1",{"Id","Name"});

        ArrayList<String> al = new ArrayList<String>();
        al = test.getCol(tableName,"colFamily1","Name");

        ArrayList<HashMap<String, String>> listofmaps = new ArrayList<HashMap<String, String>>();
        listofmaps = test.getTable(tableName,colFamilyNames,colNames);

    // specify the rowKey as per your table

    test.deleteTableRow(tableName, rowKey);
    }
}

My next post would comprise of rest of the functions I have mentioned that would hopefully help to perform almost any kind of operation on your table.
Suggestions would be welcomed !!!

Frequent Errors during Installation or Startup of Oozie Server

In spite of the fact that, Oozie Installation Steps have been entered down quite intelligibly and makes one hallucinate it as a matter of a few minutes, but as per my experience, its not that easy.. So, to somehow make it easy for you, one of my previous post is a jest of all those docs that are meant to help you install Oozie. Still here in this post, I am providing the solution to the 5 most common errors that you might have unfortunately encountered.

Error 1 :

Cannot create /var/run/oozie/oozie.pid: Directory nonexistent

Solution : 

Changing the permissions of the run folder as in

sudo chmod -cR 777 ./run 

sudo chown root:root -R /var/run/

Error 2 :

put: org.apache.hadoop.security.AccessControlException: Permission denied: user=jt, access=WRITE, inode="user":root:supergroup:rwxr-xr-x

Solution :

Add the following entry to your hadoop setup's conf/hdfs-site.xml 

<property> 

<name>dfs.permissions</name> 

<value>false</value> 

</property> 

Error 3 :

put:org.apache.hadoop.hdfs.server.namenode.SafeModeException:Cannot create directory /user/jt/examples. Name node is in safe mode.

Solution :

Use "hadoop dfsadmin -safemode leave" command to make the namenode leave safe mode forcefully. 

Or use "hadoop dfsadmin -safemode wait" to block till NN leaves by itself. 

If you need to get your cluster up and running quickly, you can manipulate the parameter dfs.namenode.threshold.percent. 

If you set it to 0, NN will not enter safe mode. 

Error 4 :

E0902: Exception occured: [java.io.IOException: Call to localhost/127.0.0.1:9000 failed on local exception: java.io.EOFException]

Solution : 

Check whether the port nos of jobtracker and namenode are correctly set in the job.properties file of the application you are running. 

Error 5 :

Hadoop StartUp Issue :  Hadoop fs command not working and datanode is not running

Solution : 

localpath_to_hadoop_data_store/dfs/data/current/VERSION and localpath_to_hadoop_data_store/dfs/name/current/VERSION  should have the same ids , if they are not change that of the datanode(s) .

If these included one of the points where you were stuck up, I hope to have helped you. All the very best for Oozie ...

Try On Oozie

While writing this post, I am assuming that you have

• Installed oozie on your linux machine
• Installed hadoop-0.20.1+

If not, I have already mentioned the Oozie Installation in my previous post.

Steps to get an Oozie app running

Having started hadoop and the oozie server, follow the steps below to get an sample oozie application running :

• In case Oozie installation has been done using debian packag, you can find the oozie examples tar.gz at /etc/oozie/doc/oozie else it can be located in the oozie setup folder.
• Extract this and the obtained /examples folder would contain apps, input-data and src sub-directories.
• Add the following properties to the conf/core-site.xml of your hadoop setup
  
  

<property>      <name>hadoop.proxyuser.oozie.hosts</name>                                                      <value>*</value>  </property>          <property>       <name>hadoop.proxyuser.oozie.groups</name>       <value>*</value>  </property>

• In order to run any of the apps, remember to edit the port nos. of jobtracker and the namenode in the job.properties file of the app depending upon your hadoop configuration

              JobTracker port no. is set in: /conf/mapred-site.xml

              and NameNode port no. is set in : /conf/core-site.xml

Accordingly replace the 'JTPortNo' and 'NNPortNo' in job.properties as below :

oozie.wf.application.path=hdfs://localhost:NNPortNo/path_to_examples/apps/map-reduce jobTracker=localhost:JTPortNo nameNode=hdfs://localhost:NNPortNo queueName=default outputDir=map-reduce

• Now its time to copy the examples dir to hdfs, but if there is already an examples dir in hdfs you must delete it else the files are not copied. Here's the command :

/path_to_hadoopdir/bin/hadoop fs -put /path_to_egdir/examples examples

For a confirmation, you can check if the copy has been successful at

        

                     http://localhost:50070

• Run the following command to get the example running

In case of debian package used for installation of Oozie

/usr/lib/oozie/bin/oozie job -oozie http://localhost:11000/oozie -config /path_to_egdir/examples/apps/map-reduce/job.properties -run

                                                                           else

/path_to_oozie/bin/oozie job -oozie http://localhost:11000/oozie -config  /path_to_egdir/examples/apps/map-reduce/job.properties -run

Here an important note is that you need to specify the local system path to job.properties and not that of hdfs in the command.

If the application has started off successfully, a job id would be returned in response to the above command, something like this :

                    job: 14-20090525161321-oozie-tucu 

• If you have the web console installed, you can view the status of the job on 

                  http://localhost:11000/oozie

else the following command will do

/path_to_oozie/bin/oozie job -oozie http://localhost:11000/oozie -info 14-20090525161321-oozie-tucu

That's it … You can apply the same steps for running any of the documented examples. Well, if the things have not worked as smoothly as they seem, my next post on 'Errors while installation and running Oozie' could be an answer.

Oozie Installation

This post is an attempt to provide you with a very consolidated list of commands that are required to install Oozie, since the documentation involves many optional steps and requires you to refine lot many links to get to correct procedure.

As mentioned in my previous post, Oozie has two flavors, one is the Cloudera distribution and the other is the Yahoo distribution. This post is an attempt to state Cloudera Oozie's installation using two methods :

• Installing Oozie debian package
• Installing Oozie tarball 

Installing Cloudera's Oozie

Prerequisites

• A Unix-like system (tested on Centos 5.5, Ubuntu 9.10+, SUSE Linux Enterprise Server 11, OS X 10.6)
• Java 1.6+ (tested with JRE 1.6.0_20)
• A Unix user and group named oozie on your machine

Installing Oozie debian package

The Oozie debian package for installing Oozie has separate packages for Oozie server(oozie) and the client(oozie-client). 

  1. Download them from the following link :

• http://archive.cloudera.com/debian/pool/contrib/o/

  
  2. Before proceeding further, it is required to install the Cloudera's Yum, for which you need to add a repository by creating a new file (a normal text file named cloudera.list) /etc/apt/sources.list.d/cloudera.list with the following two lines of content.

• deb http://archive.cloudera.com/debian <RELEASE>-cdh3 contrib
• deb-src http://archive.cloudera.com/debian <RELEASE>-cdh3 contrib

where <RELEASE> is to be replaced by the output of the command lsb_release -c.

  
  3. Run the following command

$ sudo apt-get update

  4. After this simply install the oozie server and client using the debian packages.  

  5. Start the oozie server using

• $ sudo -u oozie /usr/lib/oozie/bin/oozie-start.sh

  6. Stop the oozie server

• $ sudo -u oozie /usr/lib/oozie/bin/oozie-stop.sh

Installing Oozie Tarball

  1. Download the Oozie tarball(ver Oozie 2.3.0+31.2) from :

• https://ccp.cloudera.com/display/SUPPORT/CDH3+Downloadable+Tarballs

  
  2. Unpack the tarball in the appropriate directory. For example: 

• $ (cd /home/abc/ && sudo tar -zxvf <PATH_TO_OOZIE_TAR_GZ>)

  
  3. Change ownership of the Oozie installation to oozie:oozie:

• $ sudo chown -R oozie:oozie /home/abc/oozie-2.3.0+31.2

This installs both the client and the server and the directory contains all the necessary client and server files. 

  
  4. Start the oozie server

• $ sudo -u oozie /home/abc/oozie-*/bin/oozie-start.sh

  5. Stop the oozie server

• $ sudo -u oozie /home/abc/oozie-*/bin/oozie-stop.sh

Oozie Web Console

Though it is optional to use the web console, and there are command line utilities that can be used instead, I would recommend installing it as it presents a very clear picture of the running oozie jobs and is very easy to install.

  1. Download the ExtJS library from

• http://extjs.com/deploy/ext-2.2.zip

  
  2. Place it in a convinient location and add to Oozie through this command

• $ sudo -u oozie /usr/lib/oozie/bin/oozie-setup.sh -extjs /path_to_ExtJS/ext-2.2.z

                          or 

• $ sudo -u oozie /home/abc/oozie-*/bin/oozie-setup.sh -extjs /path_to_ExtJS/ext-2.2.zip

  
  3. Having done this, after starting the oozie server as described above , you can view the console at :

• http://localhost:11000/oozie

That's all about installing Oozie, my next post would be on how to run an oozie sample application.