Monitoring HBase using Ganglia

Configuring HBase to start emitting metrics to Ganglia is a pretty simple task. Although we can make almost (that’s for being on the safer side) any bigdata technology report  its metrics to Ganglia using JMX, but some of the technologies like HBase, Hadoop, Kafka etc. have got in-built support for reporting their metrics to Ganglia, which makes our task much more easier.

For obvious reasons, it’s essential to have a Ganglia cluster before you can start to report HBase metrics on to it and browse it's UI. For details on the components of Ganglia and how to install it, you can access the following link:

http://sourceforge.net/apps/trac/ganglia/wiki/Ganglia%203.1.x%20Installation%20and%20Configuration

Setup gmond servers on all the nodes of your hbase cluster which you want to emit metrics/whose metrics you would be interested to see on the Ganglia Web UI.

Configurations:

Once, you have Ganglia gmond servers setup on all the hbase cluster nodes being considered for reporting, follow the steps below on all the nodes of your HBase cluster

• Edit /hbase_setup/conf/hadoop-metrics.properties on all the nodes of the hbase cluster and add the following lines to it. 

/hbase-setup/conf/hadoop-metrics.properties

hbase.class=org.apache.hadoop.metrics.ganglia.GangliaContext31 hbase.period=10 hbase.servers=master.node.IP.address:8649

Note: If the version of Ganglia installed is 3.1.x the hbase.class would be assigned org.apache.hadoop.metrics.ganglia.GangliaContext31 and if the Ganglia version is older than this, it's value would be org.apache.hadoop.metrics.ganglia.GangliaContext

Next it’s time to let Ganglia metad server know that your HBase cluster nodes would be reporting their metrics to it. For this, edit /etc/ganglia/gmetad.conf(or where ever the gmetad.conf file is located) file by specifying the data_source in the configuration. The format of the data_source line is as follows:

data_source "hbase-cluster" gmetad_server_ip_address:8655

You can provide any name to your cluster, which would replace hbase-cluster in the above entry and “gmetad_server_ip_address” needs to be replaced by the IP address of your gmetad server.

That’s it. Quite easy right? Now, just advance further by restarting your hbase cluster and check out the Ganglia UI for the HBase cluster nodes metrics.

Adding/Removing RegionServers dynamically from a Fully-Distributed HBase cluster

Adding a RegionServer dynamically to a running HBase Cluster

Being able to add RegionServers dynamically without bringing down the entire cluster is quite a significant requirement and at the same time damn easy.

Step 1: Edit $HBASE_HOME/conf/regionservers on the Master node and add the new address.

Step 2: Setup the new node with needed software, permissions.

Step 3: On that node run

$HBASE_HOME/bin/hbase-daemon.sh start regionserver

Confirm it worked by looking at the Master's web UI or in that region server's log.

That’s it. You’re done.

Removing/Decommissioning a RegionServer dynamically from a running HBase Cluster

Just as significant as being able to add a RegionServer to an HBase Cluster dynamically is essential, just that important it is to be able to remove one.  

And you can be happy that it is just a matter of two steps.

Step 1: On the node to be removed run

$HBASE_HOME/bin/hbase-daemon.sh stop regionserver

Step 2: Edit $HBASE_HOME/conf/regionservers on the Master node and remove the address of the particular node being removed.

You can check the UI to ensure the node has been removed from the cluster.

Be careful about your data

Ensure to disable the Load Balancer before Decommissioning/Removing a node. This is because there might be contention between the Master recovering from the loss of the recently removed RegionServer and the running Load Balancer. Hence, it important to stop the load balancer(if running)before removing the regionserver from the cluster.

Also, in Apache HBase 0.90.2, there is an added facility for having a node gradually shed its load and then shutdown itself down by using the graceful_stop.sh script as follows:

$HBASE_HOME/bin/graceful_stop.sh HOSTNAME

where HOSTNAME is the host carrying the RegionServer you would remove/decommission.

Usual Errors encountered when using HBase

Error 1: Error on the hbase shell

Exception in thread "main" java.lang.NoClassDefFoundError: org/jruby/Main
Caused by: java.lang.ClassNotFoundException: org.jruby.Main
      at java.net.URLClassLoader$1.run(URLClassLoader.java:217)
      at java.security.AccessController.doPrivileged(Native Method)
      at java.net.URLClassLoader.findClass(URLClassLoader.java:205)
      at java.lang.ClassLoader.loadClass(ClassLoader.java:321)
      at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:294)
      at java.lang.ClassLoader.loadClass(ClassLoader.java:266)
Could not find the main class: org.jruby.Main. Program will exit.

Solution: This is most likely a version incompatibility issue, which I had encountered when trying to run hbase-0.20.4 with hadoop 0.20.1

The same setup of Hadoop was working fine with hbase-0.20.1, hbase-0.20.2 and hbase-0.20.3. Hence try with some other version of either Hadoop or HBase.

Error 2: Error on the hbase shell
Exception in thread "main" java.lang.NoClassDefFoundError: org/apache/hadoop/hbase/zookeeper/ZKServerTool
Caused by: java.lang.ClassNotFoundException: org.apache.hadoop.hbase.zookeeper.ZKServerTool
      at java.net.URLClassLoader$1.run(URLClassLoader.java:217)
      at java.security.AccessController.doPrivileged(Native Method)
      at java.net.URLClassLoader.findClass(URLClassLoader.java:205)
      at java.lang.ClassLoader.loadClass(ClassLoader.java:321)
      at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:294)
      at java.lang.ClassLoader.loadClass(ClassLoader.java:266)
Could not find the main class: org.apache.hadoop.hbase.zookeeper.ZKServerTool. Program will exit.
 

Solution: Same as that for Error 1 above

Error 3: Error in the master logs
ERROR org.apache.hadoop.hbase.master.HMaster: Cannot start master
Caused by: java.net.ConnectException: Call to hadoop-4252-nameserver/192.168.41.123:9000 failed on connection exception: java.net.ConnectException: Connection refused

Solution: When getting the above error, my /etc/hosts file on the HBase Master node was something like below:

127.0.0.1   hbase-master      localhost.localdomain   localhost
::1   hbase-master      localhost6.localdomain6 localhost6
192.168.41.123  hbase-master
192.168.41.123  localhost
192.168.41.123  localhost.localdomain

And to resolve the error, I edited it to be

192.168.41.123  hbase-master
192.168.41.123  localhost
192.168.41.123  localhost.localdomain

After which the above error went away.

Error 4: Error in the master logs
ERROR org.apache.hadoop.hbase.master.HMasterCommandLine: Failed to start master
java.lang.RuntimeException: Failed construction of Master: class org.apache.hadoop.hbase.master.HMaster
      at org.apache.hadoop.hbase.master.HMaster.constructMaster(HMaster.java:1105)
      at org.apache.hadoop.hbase.master.HMasterCommandLine.startMaster(HMasterCommandLine.java:142)
      at org.apache.hadoop.hbase.master.HMasterCommandLine.run(HMasterCommandLine.java:102)
      at org.apache.hadoop.util.ToolRunner.run(ToolRunner.java:65)
      at org.apache.hadoop.hbase.util.ServerCommandLine.doMain(ServerCommandLine.java:76)
      at org.apache.hadoop.hbase.master.HMaster.main(HMaster.java:1119)
Caused by: java.lang.ClassNotFoundException: org.apache.commons.configuration.Configuration
      at java.net.URLClassLoader$1.run(URLClassLoader.java:217)
      at java.security.AccessController.doPrivileged(Native Method)
      at java.net.URLClassLoader.findClass(URLClassLoader.java:205)
      at java.lang.ClassLoader.loadClass(ClassLoader.java:321)
      at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:294)
      at java.lang.ClassLoader.loadClass(ClassLoader.java:266)
      at org.apache.hadoop.metrics2.lib.DefaultMetricsSystem.<init>(DefaultMetricsSystem.java:37)
      at org.apache.hadoop.metrics2.lib.DefaultMetricsSystem.<clinit>(DefaultMetricsSystem.java:34)
      at org.apache.hadoop.security.UgiInstrumentation.create(UgiInstrumentation.java:51)
      at org.apache.hadoop.security.UserGroupInformation.initialize(UserGroupInformation.java:216)
      at org.apache.hadoop.security.UserGroupInformation.ensureInitialized(UserGroupInformation.java:184)
      at org.apache.hadoop.security.UserGroupInformation.isSecurityEnabled(UserGroupInformation.java:236)
      at org.apache.hadoop.security.KerberosName.<clinit>(KerberosName.java:79)
      at org.apache.hadoop.security.UserGroupInformation.initialize(UserGroupInformation.java:209)
      at org.apache.hadoop.security.UserGroupInformation.ensureInitialized(UserGroupInformation.java:184)
      at org.apache.hadoop.security.UserGroupInformation.isSecurityEnabled(UserGroupInformation.java:236)
      at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
      at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
      at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
      at java.lang.reflect.Method.invoke(Method.java:616)
      at org.apache.hadoop.hbase.util.Methods.call(Methods.java:37)
      at org.apache.hadoop.hbase.security.User.call(User.java:457)
      at org.apache.hadoop.hbase.security.User.callStatic(User.java:447)
      at org.apache.hadoop.hbase.security.User.access$200(User.java:49)
      at org.apache.hadoop.hbase.security.User$SecureHadoopUser.isSecurityEnabled(User.java:435)
      at org.apache.hadoop.hbase.security.User$SecureHadoopUser.login(User.java:406)
      at org.apache.hadoop.hbase.security.User.login(User.java:146)
      at org.apache.hadoop.hbase.master.HMaster.<init>(HMaster.java:204)
      at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
      at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:57)
      at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
      at java.lang.reflect.Constructor.newInstance(Constructor.java:532)
      at org.apache.hadoop.hbase.master.HMaster.constructMaster(HMaster.java:1100)
      ... 5 more

Solution: Remove the commons-configuration-*.jar(if present) from your hbase setup lib folder and replace with the commons-configuration-*.jar of the hadoop being used.
 

Error 5: Error in the master logs
java.io.IOException: Call to /10.0.1.37:50070 failed on local exception: java.io.EOFException at org.apache.hadoop.ipc.Client.wrapException(Client.java:1139) at org.apache.hadoop.ipc.Client.call(Client.java:1107) at org.apache.hadoop.ipc.RPC$Invoker.invoke(RPC.java:226) ....

Solution: Remove the hadoop-core-*.jar from your hbase setup lib folder and replace with the hadoop-core jar of the hadoop being used.
 

Error 6: Error in the master logs
FATAL org.apache.hadoop.hbase.master.HMaster: Unhandled exception. Starting shutdown.
java.lang.IllegalArgumentException: � 13955@hbase-masterVMENABLED
      at org.apache.hadoop.hbase.zookeeper.ZKTable.getTableState(ZKTable.java:119)
      at org.apache.hadoop.hbase.zookeeper.ZKTable.populateTableStates(ZKTable.java:94)
      at org.apache.hadoop.hbase.zookeeper.ZKTable.<init>(ZKTable.java:79)
      at org.apache.hadoop.hbase.master.AssignmentManager.<init>(AssignmentManager.java:171)
      at org.apache.hadoop.hbase.master.HMaster.finishInitialization(HMaster.java:353)
      at org.apache.hadoop.hbase.master.HMaster.run(HMaster.java:281)
2012-11-15 19:03:12,275 INFO org.apache.hadoop.hbase.master.HMaster: Aborting
2012-11-15 19:03:12,277 DEBUG org.apache.hadoop.hbase.master.HMaster: Stopping service threads
2012-11-15 19:03:12,279 INFO org.apache.hadoop.ipc.HBaseServer: Stopping server on 60000
2012-11-15 19:03:12,292 INFO org.apache.hadoop.ipc.HBaseServer: IPC Server handler 0 on 60000: exiting
2012-11-15 19:03:12,296 INFO org.apache.hadoop.ipc.HBaseServer: IPC Server handler 1 on 60000: exiting
2012-11-15 19:03:12,301 INFO org.apache.hadoop.ipc.HBaseServer: IPC Server handler 2 on 60000: exiting
2012-11-15 19:03:12,303 INFO org.apache.hadoop.ipc.HBaseServer: IPC Server handler 3 on 60000: exiting
2012-11-15 19:03:12,306 INFO org.apache.hadoop.ipc.HBaseServer: IPC Server handler 4 on 60000: exiting
2012-11-15 19:03:12,308 INFO org.apache.hadoop.ipc.HBaseServer: IPC Server handler 5 on 60000: exiting
2012-11-15 19:03:12,315 INFO org.apache.hadoop.ipc.HBaseServer: IPC Server handler 6 on 60000: exiting
2012-11-15 19:03:12,318 INFO org.apache.hadoop.ipc.HBaseServer: IPC Server handler 7 on 60000: exiting
2012-11-15 19:03:12,321 INFO org.apache.hadoop.ipc.HBaseServer: IPC Server handler 8 on 60000: exiting
2012-11-15 19:03:12,322 INFO org.apache.hadoop.ipc.HBaseServer: IPC Server handler 9 on 60000: exiting
2012-11-15 19:03:12,329 INFO org.apache.hadoop.ipc.HBaseServer: Stopping IPC Server listener on 60000
2012-11-15 19:03:12,361 INFO org.apache.hadoop.ipc.HBaseServer: Stopping IPC Server Responder
2012-11-15 19:03:12,471 DEBUG org.apache.hadoop.hbase.catalog.CatalogTracker: Stopping catalog tracker org.apache.hadoop.hbase.catalog.CatalogTracker@73305c
2012-11-15 19:03:12,472 DEBUG org.apache.hadoop.hbase.client.HConnectionManager$HConnectionImplementation: The connection to hconnection-0x13b0445c6c90001 has been closed.
2012-11-15 19:03:12,473 INFO org.apache.hadoop.hbase.client.HConnectionManager$HConnectionImplementation: Closed zookeeper sessionid=0x13b0445c6c90001
2012-11-15 19:03:12,498 INFO org.apache.zookeeper.ZooKeeper: Session: 0x13b0445c6c90001 closed
2012-11-15 19:03:12,499 DEBUG org.apache.hadoop.hbase.client.HConnectionManager$HConnectionImplementation: The connection to null has been closed.
2012-11-15 19:03:12,499 INFO org.apache.zookeeper.ClientCnxn: EventThread shut down
2012-11-15 19:03:12,532 INFO org.apache.zookeeper.ZooKeeper: Session: 0x13b0445c6c90000 closed
2012-11-15 19:03:12,532 INFO org.apache.hadoop.hbase.master.HMaster: HMaster main thread exiting
2012-11-15 19:03:12,533 INFO org.apache.zookeeper.ClientCnxn: EventThread shut down

Solution: Remove the commons-lang-*.jar from your hbase setup lib folder and replace with the commons-lang-*.jar of the hadoop being used.
 

HBase Installation : Fully Distributed Mode

HBase can be installed in 3 modes:

1. Standalone Mode
2. Pseudo-Distributed Mode
3. Fully-Distributed Mode

The objective of this post is to provide a tried-and-true procedure for installing HBase in Fully-Distributed Mode. But before we dive deeper into installation nuts and bolts, here are some hbase preliminaries, that I feel I should include as a startup.

1. When talking about installing HBase in fully distributed mode we'll be addressing the following:

• HDFS: A running instance of HDFS is required for deploying HBase in distributed mode.
• HBase Master: HBase cluster has a master-slave architecture where the HBase Master is responsible for monitoring all the slaves i.e. Region Servers.
• Region Servers: These are the slave nodes responsible for storing and managing regions.
• Zookeeper Cluster: A distributed Apache HBase installation depends on a running ZooKeeper cluster. All participating nodes and clients need to be able to access the running ZooKeeper ensemble.

2. The coin of setting up a Fully Distributed HBase Cluster has got two sides to it:

• When Zookeeper cluster is managed by HBase internally
• When Zookeeper cluster is managed externally

3. HBase is overparticular about the DNS entries of its cluster nodes. Therefore, to avert imminent discrepancies we would be assigning host names to the cluster nodes and using them for installation.

Deploying a Fully-Distributed HBase Cluster

Assumptions

For the purpose of clarity and ease of expression, I'll be assuming that we are setting up a cluster of 3 nodes with IP Addresses
10.10.10.1
10.10.10.2
10.10.10.3
where 10.10.10.1 would be the master and 10.10.10.2,3 would be the slaves/region servers.
Also, we'll be assuming that we have a running instance of HDFS, whose NameNode daemon is running on
10.10.10.4

Case I- When HBase manages the Zookeeper ensemble
HBase by default manages a ZooKeeper "cluster" for you. It will start and stop the ZooKeeper ensemble as part of the HBase start/stop process.

Step 1: Assign hostnames to all the nodes of the cluster.

10.10.10.1 master 10.10.10.2 regionserver1 10.10.10.3 regionserver2 10.10.10.4 namenode

Now in each of these append the required hostnames to the /etc/hosts file.

On the Namenode(10.10.10.4) add:

10.10.10.4 namenode

On the Master Node(10.10.10.1) add:

10.10.10.1 master 10.10.10.4 namenode 10.10.10.2 regionserver1 10.10.10.3 regionserver2

On the Region Server 1(10.10.10.2) add:

10.10.10.1 master 10.10.10.2 regionserver1

And on the Region Server 2(10.10.10.3) add:

10.10.10.1 master 10.10.10.3 regionserver2

Step 2: Download a stable release of hbase from
http://apache.techartifact.com/mirror/hbase/
and untar it at a suitable location on all the hbase cluster nodes(10.10.10.1, 10.10.10.2, 10.10.10.3).

Step 3: Edit the /conf/hbase-env.sh file on all the hbase cluster nodes(10.10.10.1, 10.10.10.2, 10.10.10.3) to add the JAVA_HOME (for eg. /usr/lib/jvm/java-6-openjdk/) and to set the HBASE_MANAGES_ZK to true to indicate that HBase is supposed to manage the zookeeper ensemble internally.

export JAVA_HOME=your_java_home export HBASE_MANAGES_ZK=true

Step 4: Edit the /conf/hbase-site.xml on all the hbase cluster nodes which after your editing should look like:

<configuration> <property>     <name>hbase.master</name>     <value>10.10.10.1:60000</value> </property> <property>     <name>hbase.rootdir</name>     <value>hdfs://namenode:9000/hbase</value> </property> <property>     <name>hbase.cluster.distributed</name>     <value>true</value> </property> </configuration>

Here the property 'hbase.master' reflects the host and port that the HBase master(10.10.10.1) runs at. Next is the 'hbase.rootdir' property which is a directory shared by the region servers. The value has to be an HDFS location, for eg: hdfs://namenode:9000/hbase. Since we have assigned 10.10.10.4, hostname 'namenode' whose NameNode port is assumed to be 9000, we form the rootdir location as hdfs://namenode:9000/hbase. If your namenode is running on some other port replace 9000 by that port number.

Step 5: Edit the /conf/regionservers file on all the hbase cluster nodes. Add the hostnames of all the region server nodes. For eg.

regionserver1 regionserver2

This completes the installation process of HBase Cluster with Zookeeper Ensemble being managed internally.

Case II- When the Zookeeper ensemble is managed externally
We can manage the ZooKeeper ensemble independent of HBase and just point HBase at the cluster it should use.

Step 1: Assign hostnames to all the nodes of the hbase and zookeeper cluster. Assuming we have a two node zookeeper cluster of nodes 10.10.10.5 and 10.10.10.6

10.10.10.1 master 10.10.10.2 regionserver1 10.10.10.3 regionserver2 10.10.10.4 namenode 10.10.10.5 zkserver1 10.10.10.6 zkserver2

Now in each of these append the required hostnames to the /etc/hosts file.
On the Namenode(10.10.10.4) add:

10.10.10.4 namenode

On the Master Node(10.10.10.1) add:

10.10.10.1 master 10.10.10.4 namenode 10.10.10.2 regionserver1 10.10.10.3 regionserver2 10.10.10.5 zkserver1 10.10.10.6 zkserver2

On the Region Server 1(10.10.10.2) add:

10.10.10.1 master 10.10.10.2 regionserver1 10.10.10.5 zkserver1 10.10.10.6 zkserver2

And on the Region Server 2(10.10.10.3) add:

10.10.10.1 master 10.10.10.3 regionserver2 10.10.10.5 zkserver1 10.10.10.6 zkserver2

Step 2: Download a stable release of hbase from
http://apache.techartifact.com/mirror/hbase/
and untar it at a suitable location on all the hbase cluster nodes(10.10.10.1, 10.10.10.2, 10.10.10.3).

Step 3: Edit the /conf/hbase-env.sh file on all the hbase cluster nodes(10.10.10.1, 10.10.10.2, 10.10.10.3) to add the JAVA_HOME (for eg. /usr/lib/jvm/java-6-openjdk/) and to set the HBASE_MANAGES_ZK to false to indicate that the zookeeper ensemble would be managed externally.

export JAVA_HOME=your_java_home export HBASE_MANAGES_ZK=false

Step 4: Edit the /conf/hbase-site.xml on all the hbase cluster nodes which after editing should look like:

<configuration> <property>     <name>hbase.master</name>     <value>10.10.10.1:60000</value> </property> <property>     <name>hbase.rootdir</name>     <value>hdfs://namenode:9000/hbase</value> </property> <property>     <name>hbase.cluster.distributed</name>     <value>true</value> </property> <property>     <name>hbase.zookeeper.property.clientPort</name>     <value>2181</value>  </property> <property>       <name>hbase.zookeeper.quorum</name>       <value>zkserver1,zkserver2</value> </property> </configuration>

Here the property 'hbase.master' reflects the host and port that the HBase master(10.10.10.1) runs at.
Next is the 'hbase.rootdir' property which is a directory shared by the region servers. The value has to be an HDFS location, for eg: hdfs://namenode:9000/hbase. Since we have assigned 10.10.10.4, hostname 'namenode' whose NameNode port is assumed to be 9000, we form the rootdir location as hdfs://namenode:9000/hbase. If your namenode is running on some other port replace 9000 by that port number.
The property 'hbase.zookeeper.property.clientPort' reflects a property from ZooKeeper's config zoo.cfg. It is the port at which the clients will connect.
And lastly the property 'hbase.zookeeper.quorum' is a comma separated list of servers in the ZooKeeper Quorum.

Step 5: Edit the /conf/regionservers file on all the hbase cluster nodes. Add the hostnames of all the region server nodes. For eg.

regionserver1 regionserver2

This brings us to an end of the installation process for HBase cluster with externally managed zookeeper ensemble.

Start your HBase Cluster
Having followed the steps above, now its time to start the deployed cluster. If your have an externally managed zookeeper cluster, make sure to start it before you proceed further.
On the master node(10.10.10.1) cd to the hbase setup and run the following command

$HBASE_HOME/bin/start-hbase.sh

This would start all the master and the region servers on respective nodes of the cluster.

Stop your HBase Cluster
To stop a running cluster, on the master node, cd to the hbase setup and run

$HBASE_HOME/bin/stop-hbase.sh

Hadoop-HBase Version Compatibility
As per my assessment, hadoop-1.0.4 and hadoop-1.0.3 versions of Hadoop work fine with all hbase versions of the series 0.94.x and 0.92.x, but not the 0.90.x series and the older releases due to version incompatibility.

Storm-HBase Integration : Pour the Data Stream into HBase

Storm is proclaimed to be the “Hadoop of Realtime Processing” and it’s quite justified to be calling it so. Hadoop is solely meant for batch processing of enormous amount of data and on the other end Storm is built for processing realtime data. Where Hadoop uses Map Reduce jobs for the batch processing of large sets of distributed data, Storm runs topologies which are a chain of heterogeneous Storm elements (spouts and bolts)  for processing the realtime data stream.

Hadoop utilizes HBase for storing this large amount of data in a somewhat structured way. And to what I realize, it is essential for Storm to have an equally potential and high capacity storage where the running topology can pour in the processed data.

Why not HBase be this “potential and high capacity storage” then?

Though I don’t claim to have all the knowledge in the universe about all these complex systems, but Yes !!! HBase can definitely be the pick mainly due to giant storage capacity it possesses.

In an attempt to justify my big “Yes !!! “ I tried out this integration of Storm and HBase the outcome of which is a Storm Bolt that can dump the stream into HBase with just a few obvious inputs.


Requirements :

• Setup Hadoop locally on your machine (tested version 0.20.2-cdh3u2)
• Setup HBase locally on your  machine (tested version 0.90.4-cdh3u2)
• Setup Storm locally on  your machine (tested version 0.7.1)

Following is the source code of the Bolt that delivers the stream data into HBase

HBase Dumper Bolt :
This bolt expects the hbase table name, list of column family names and respective column names, and a variable "rowKeyCheck" which is supposed to be set to the column name whose value is the be used as the row key and if timestamp is to be used the "rowKeyCheck" should be "timestamp".
Every time it receives a tuple, it creates a list of values accordingly and calls a method of HBaseCommunicator class which dumps it into the hbase table.

import java.sql.Timestamp; import java.util.ArrayList; import java.util.Date; import java.util.Map; import org.apache.hadoop.hbase.HBaseConfiguration; import backtype.storm.task.OutputCollector; import backtype.storm.task.TopologyContext; import backtype.storm.topology.BasicOutputCollector; import backtype.storm.topology.IBasicBolt; import backtype.storm.topology.OutputFieldsDeclarer; import backtype.storm.tuple.Tuple; /*  * Bolt for dumping stream data into hbase  */ public class HBaseDumperBolt implements IBasicBolt {     private static final long serialVersionUID = 1L;     private static transient HBaseConnector connector = null;     private static transient HBaseConfiguration conf = null;     private static transient HBaseCommunicator communicator = null;     private OutputCollector _collector;     private Date today = null;     private Timestamp timestamp = null;     private ArrayList<String> colFamilyNames = new ArrayList<String>();     private ArrayList<ArrayList<String>>  colNames = new ArrayList<ArrayList<String>>();     private ArrayList<ArrayList<String>> colValues = new ArrayList<ArrayList<String>>();     private ArrayList<String> colFamilyValues = new ArrayList<String>();     private String rowKeyCheck = null, rowKey = null, fieldValue = null, tableName = null;;     private static int counter = 0;        private long time;     public HBaseDumperBolt(final String hbaseXmlLocation, final String tableName, final String rowKeyCheck, final ArrayList<String> colFamilyNames, final ArrayList<ArrayList<String>> colNames) {         this.tableName = tableName;         this.colFamilyNames = colFamilyNames;         this.colNames = colNames;         this.rowKeyCheck = rowKeyCheck;         connector = new HBaseConnector();         conf = connector.getHBaseConf(hbaseXmlLocation);         communicator = new HBaseCommunicator(conf);         //check if tableName already exists         if (colFamilyNames.size() == colNames.size()) {             if (!communicator.tableExists(tableName)) {                 communicator.createTable(tableName, colFamilyNames);             }         }     }     public void execute(Tuple tuple, BasicOutputCollector collector) {         counter = 0;         rowKey = null;         colValues = new ArrayList<ArrayList<String>>();         if (colFamilyNames.size() == 1) {             for (int j = 0; j < colNames.get(0).size(); j++) {                 fieldValue = tuple.getValue(j).toString();                 if (rowKeyCheck.equals(colNames.get(0).get(j))) {                     rowKey = fieldValue;                 }                 colFamilyValues.add(fieldValue);             }             colValues.add(colFamilyValues);         } else {             for (int i = 0; i < colFamilyNames.size(); i++) {                 for (int j = 0; j < colNames.get(i).size(); j++) {                     fieldValue = tuple.getValue(counter).toString();                     if (rowKeyCheck.equals(colNames.get(i).get(j))) {                         rowKey = fieldValue;                     }                     colFamilyValues.add(fieldValue);                     counter++;                 }                 colValues.add(colFamilyValues);                 colFamilyValues = new ArrayList<String>();             }         }         if (rowKeyCheck.equals("timestamp") && rowKey == null) {             today = new Date();             timestamp = new Timestamp(today.getTime());             time = timestamp.getTime();             rowKey = String.valueOf(time);         }         communicator.addRow(rowKey, tableName, colFamilyNames, colNames, colValues);     }     public void prepare(Map confMap, TopologyContext context,             OutputCollector collector) {         _collector = collector;     }     public void cleanup() {     }     public void declareOutputFields(OutputFieldsDeclarer declarer) {     }     @Override     public Map<String, Object> getComponentConfiguration() {         Map<String, Object> map = null;         return map;     }     @Override     public void prepare(Map stormConf, TopologyContext context) {     } }

The supporting classes required for the bolt are HBaseCommunicator.java and HBaseConnector.java.

HBase Communicator :

This class defines all the functions needed to communicate with hbase database.

import java.io.IOException; import java.util.ArrayList; import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.HColumnDescriptor; import org.apache.hadoop.hbase.HTableDescriptor; import org.apache.hadoop.hbase.client.Get; import org.apache.hadoop.hbase.client.HBaseAdmin; 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.util.Bytes; /*  * This class implements methods for checking an hbase table's existence, creating an hbase table and inserting a row into the hbase table  */ public class HBaseCommunicator {     private static HBaseConfiguration conf;     private HTable table = null;     private Put putdata = null;     private HBaseAdmin admin = null;     private String colFamilyName = null, colValue = null, result = null;     private byte[] rowKeyBytes = null, key = null, columnValue = null;     private Get getRowObj = null;     private Result rowEntries = null;                    public HBaseCommunicator(final HBaseConfiguration conf)    {         this.conf = conf;     }     // check if the table exists     public final boolean tableExists(final String tableName) {         try {             admin = new HBaseAdmin(conf);             if (admin.tableExists(tableName)) {                 return true;             }         } catch (Exception e) {             System.out.println("Exception occured while checking table's existence");             e.printStackTrace();         }         return false;     }    // creates a table     public final void createTable(final String tableName, final ArrayList<String> colFamilies) {         try {             HBaseAdmin hbase = new HBaseAdmin(conf);             HTableDescriptor desc = new HTableDescriptor(tableName);             for(int i = 0; i < colFamilies.size(); i++) {                 HColumnDescriptor meta = new HColumnDescriptor(colFamilies.get(i).getBytes());                 desc.addFamily(meta);             }             hbase.createTable(desc);         } catch (Exception e) {             System.out.println("Exception occured creating table in hbase");             e.printStackTrace();         }     }  // add row to a table     public final void addRow(final String rowKey, final String tableName, final ArrayList<String> colFamilies, final ArrayList<ArrayList<String>> colNames, final ArrayList<ArrayList<String>> data) {         try    {             colFamilyName = null;             rowKeyBytes = null;             putdata = null;             table = new HTable(conf, tableName);             //rowKey = "row" + (int)(Math.random() * 1000);             rowKeyBytes = Bytes.toBytes(rowKey);             putdata = new Put(rowKeyBytes);             for (int i = 0 ; i < colFamilies.size(); i++) {                 colFamilyName = colFamilies.get(i);                 if (colNames.get(i).size() == data.get(i).size())                 {                     for (int j = 0 ; j < colNames.get(i).size(); j++) {                         colValue = data.get(i).get(j);                         if (colValue.equals(null))                             colValue = "null";                         putdata.add(Bytes.toBytes(colFamilyName), Bytes.toBytes(colNames.get(i).get(j)),                                 Bytes.toBytes(colValue));                     }                     table.put(putdata);                 }             }         } catch (IOException e)    {             System.out.println("Exception occured in adding data");         }     }     public final String getColEntry(String tableName, String rowKey,             String colFamilyName, String colName) {         result = null;         try {             HTable table = new HTable(conf, tableName);             key = Bytes.toBytes(rowKey);             getRowObj = new Get(key);             rowEntries = table.get(getRowObj);             columnValue = rowEntries.getValue(Bytes.toBytes(colFamilyName),                     Bytes.toBytes(colName));             result = Bytes.toString(columnValue);         } catch (IOException e) {             System.out.println("Exception occured in retrieving data");         }         return result;     } }

HBase Connector : 
This class establishes a connection with the hbase database and returns a HBaseConfiguration object.

import org.apache.hadoop.fs.Path; import org.apache.hadoop.hbase.HBaseConfiguration; /*  * Class that establishes a connection with hbase and returns an HBaseConfiguration object                  */ public class HBaseConnector {     private HBaseConfiguration conf;     public final HBaseConfiguration getHBaseConf(final String pathToHBaseXMLFile) {         conf = new HBaseConfiguration();         conf.addResource(new Path(pathToHBaseXMLFile));         return conf;     } }

Feeder Spout :
A sample spout which emitting tuples of the form [String, Integer] to be dumped into hbase table.

import java.util.Map; import java.util.Random;                                              import backtype.storm.spout.SpoutOutputCollector; import backtype.storm.task.TopologyContext; import backtype.storm.topology.IRichSpout; import backtype.storm.topology.OutputFieldsDeclarer; import backtype.storm.tuple.Fields; import backtype.storm.tuple.Values; import backtype.storm.utils.Utils; public class ExclIntegerSpout implements IRichSpout {     SpoutOutputCollector _collector;     Random _rand;      int count = 0;     public boolean isDistributed() {         return true;     }     @Override     public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {                        _collector = collector;         _rand = new Random();     }     @Override     public void nextTuple() {         Utils.sleep(1000);         String[] words = new String[] { "hello", "today", "divine", "optimized", "canon"};         Integer[] numbers = new Integer[] {                 1,2,3,4,5         };         if(count == numbers.length -1) {             count = 0;         }         count ++;         int number = numbers[count];         String word = words[count];         int randomNum = (int) (Math.random()*1000);         System.out.println("Random Number: " +randomNum);         System.out.println("ExclSpout emitting : " + number);         _collector.emit(new Values(word, number));               }     @Override     public void close() {            }     @Override     public void ack(Object id) {     }      @Override     public void fail(Object id) {     }     @Override     public void declareOutputFields(OutputFieldsDeclarer declarer) {         declarer.declare(new Fields("word", "number"));     }     @Override     public void activate() {}     @Override     public void deactivate() {}     @Override     public Map<String, Object> getComponentConfiguration() {         return null;     } }

Here’s a sample topology showcasing the use of the bolt.

Topology implemeting the HBase Bolt :

import java.util.ArrayList; import java.util.HashMap; import java.util.Map; import backtype.storm.Config; import backtype.storm.LocalCluster; import backtype.storm.topology.TopologyBuilder; import backtype.storm.utils.Utils; /*  * Topology that creates an hbase table (if not already created) with a single column family, row key being the timestamp and dumps the stream data into it.  */ public class HBaseTopology {     public static void main(String[] args) {         /*          * List of column family names of the hbase table          */         ArrayList<String> columnFamilyNames = new ArrayList<String>();         /*          * List containing list of column names per column family of the hbase table          */         ArrayList<ArrayList<String>> columnNamesList  = new ArrayList<ArrayList<String>>();         /*          * Temporary list used for populating the @columnNamesList          */         ArrayList<String> columnNames = new ArrayList<String>();         /*          * rowKey of the hbase table, assigned "timestamp" if none of the column's value but timestamp is to be used as the row key, else the column name          */         String rowKey = null;               /*          * hbase table name          */         String tableName = "testTable";               /*          * local fs location of the hbase-site.xml file          */         String hbaseXmlFileLocation = "/home/abc/hbase-0.20.6/conf/hbase-site.xml";         TopologyBuilder builder = new TopologyBuilder();          // add spout to the builder         builder.setSpout("spout", new ExclIntegerSpout(), 2);         // populate the column family name and column names list         columnFamilyNames.add("colFamily1");         columnNames.add("word");         columnNames.add("number");         columnNamesList.add(columnNames);         columnNames = new ArrayList<String>();         // set rowKey = "timestamp" if value of none of the columns is to be used as rowKey         rowKey = "timestamp";         // if some column's value is to be used as the row key, for e.g. number then use         / /rowKey = "number";         // add dumper bolt to the builder         HBaseDumperBolt dumperBolt = new HBaseDumperBolt(hbaseXmlFileLocation, tableName, rowKey, columnFamilyNames, columnNamesList);         builder.setBolt("dumperBolt", dumperBolt, 1).shuffleGrouping("spout");         Config conf = new Config();         conf.setDebug(true);         LocalCluster cluster = new LocalCluster();         cluster.submitTopology("sample-workflow", conf, builder.createTopology());         Utils.sleep(10000);         cluster.shutdown();     } }