Lineland

HBase File Locality in HDFS

2010-05-28T10:27:00.000-07:00

One of the more ambiguous things in Hadoop is block replication: it happens automatically and you should not have to worry about it. HBase relies on it 100% to provide the data safety as it stores its files into the distributed file system. While that works completely transparent, one of the more advanced questions asked though is how does this affect performance? This usually arises when the user starts writing MapReduce jobs against either HBase or Hadoop directly. Especially with larger data being stored in HBase, how does the system take care of placing the data close to where it is needed? This is referred to data locality and in case of HBase using the Hadoop file system (HDFS) there may be doubts how that is working.

First let's see how Hadoop handles this. The MapReduce documentation advertises the fact that tasks run close to the data they process. This is achieved by breaking up large files in HDFS into smaller chunks, or so called blocks. That is also the reason why the block size in Hadoop is much larger than you may know them from operating systems and their file systems. Default setting is 64MB, but usually 128MB is chosen, if not even larger when you are sure all your files are larger than a single block in size. Each block maps to a task run to process the contained data. That also means larger block sizes equal fewer map tasks to run as the number of mappers is driven by the number of blocks that need processing. Hadoop knows where blocks are located and runs the map tasks directly on the node that hosts it (actually one of them as replication means it has a few hosts to chose from). This is how it guarantees data locality during MapReduce.

Back to HBase. When you have arrived at that point with Hadoop and you now understand that it can process data locally you start to question how this may work with HBase. If you have read my post on HBase's storage architecture you saw that HBase simply stores files in HDFS. It does so for the actual data files (HFile) as well as its log (WAL). And if you look into the code it simply uses FileSystem.create(Path path) to create these. When you then consider two access patterns, a) direct random access and b) MapReduce scanning of tables, you wonder if care was taken that the HDFS blocks are close to where they are read by HBase.

One thing upfront, if you do not co-share your cluster with Hadoop and HBase but instead employ a separate Hadoop as well as a stand-alone HBase cluster then there is no data locality - and it can't be. That equals to running a separate MapReduce cluster where it would not be able to execute tasks directly on the datanode. It is imperative for data locality to have them running on the same cluster, Hadoop (as in the HDFS), MapReduce and HBase. End of story.

OK, you them all co-located on a single (hopefully larger) cluster? Then read on. How does Hadoop figure out where data is located as HBase accesses it. Remember the access pattern above, both go through a single piece of software called a RegionServer. Case a) uses random access patterns while b) scans all contiguous rows of a table but does so through the same API. As explained in my referenced post and mentioned above, HBase simply stores files and those get distributed as replicated blocks across all data nodes of the HDFS. Now imagine you stop HBase after saving a lot of data and restarting it subsequently. The region servers are restarted and assign a seemingly random number of regions. At this very point there is no data locality guaranteed - how could it be?

The most important factor is that HBase is not restarted frequently and that it performs house keeping on a regular basis. These so called compactions rewrite files as new data is added over time. All files in HDFS once written are immutable (for all sorts of reasons). Because of that, data is written into new files and as their number grows HBase compacts them into another set of new, consolidated files. And here is the kicker: HDFS is smart enough to put the data where it is needed! How does that work you ask? We need to take a deep dive into Hadoop's source code and see how the above FileSystem.create(Path path) that HBase uses works. We are running on HDFS here, so we are actually using DistributedFileSystem.create(Path path) which looks like this:

public FSDataOutputStream create(Path f) throws IOException {
  return create(f, true);
}

It returns a FSDataOutputStream and that is create like so:

public FSDataOutputStream create(Path f, FsPermission permission, boolean overwrite, int bufferSize, short replication, long blockSize, Progressable progress) throws IOException {
  return new FSDataOutputStream(dfs.create(getPathName(f), permission, overwrite, replication, blockSize, progress, bufferSize), statistics);
}

It uses a DFSClient instance that is the "umbilical" cord connecting the client with the NameNode:

this.dfs = new DFSClient(namenode, conf, statistics);

What is returned though is a DFSClient.DFSOutputStream instance. As you write data into the stream the DFSClient aggregates it into "packages" which are then written as blocks to the data nodes. This happens in DFSClient.DFSOutputStream.DataStreamer (please hang in there, we are close!) which runs as a daemon thread in the background. The magic unfolds now in a few hops on the stack, first in the daemon run() it gets the list of nodes to store the data on:

nodes = nextBlockOutputStream(src);

This in turn calls:

long startTime = System.currentTimeMillis();
lb = locateFollowingBlock(startTime);
block = lb.getBlock();
nodes = lb.getLocations();

We follow further down and see that locateFollowingBlocks() calls:

return namenode.addBlock(src, clientName);

Here is where it all comes together. The name node is called to add a new block and the src parameter indicates for what file, while clientName is the name of the DFSClient instance. I skip one more small method in between and show you the next bigger step involved:

public LocatedBlock getAdditionalBlock(String src, String clientName) throws IOException {
  ...
  INodeFileUnderConstruction pendingFile  = checkLease(src, clientName);
  ...
  fileLength = pendingFile.computeContentSummary().getLength();
  blockSize = pendingFile.getPreferredBlockSize();
  clientNode = pendingFile.getClientNode();
  replication = (int)pendingFile.getReplication();

  // choose targets for the new block tobe allocated.
  DatanodeDescriptor targets[] = replicator.chooseTarget(replication, clientNode, null, blockSize);
  ...
}

We are finally getting to the core of this code in the replicator.chooseTarget() call:

private DatanodeDescriptor chooseTarget(int numOfReplicas, DatanodeDescriptor writer, List<Node> excludedNodes, long blocksize, int maxNodesPerRack, List<DatanodeDescriptor> results) {
  
  if (numOfReplicas == 0 || clusterMap.getNumOfLeaves()==0) {
    return writer;
  }
  
  int numOfResults = results.size();
  boolean newBlock = (numOfResults==0);
  if (writer == null && !newBlock) {
    writer = (DatanodeDescriptor)results.get(0); 
  }
  
  try {
    switch(numOfResults) {
    case 0:
      writer = chooseLocalNode(writer, excludedNodes, blocksize, maxNodesPerRack, results);
      if (--numOfReplicas == 0) {
        break;
      }
    case 1:
      chooseRemoteRack(1, results.get(0), excludedNodes, blocksize, maxNodesPerRack, results);
      if (--numOfReplicas == 0) {
        break;
      }
    case 2:
      if (clusterMap.isOnSameRack(results.get(0), results.get(1))) {
        chooseRemoteRack(1, results.get(0), excludedNodes, blocksize, maxNodesPerRack, results);
      } else if (newBlock) {
        chooseLocalRack(results.get(1), excludedNodes, blocksize, maxNodesPerRack, results);
      } else {
        chooseLocalRack(writer, excludedNodes, blocksize, maxNodesPerRack, results);
      }
      if (--numOfReplicas == 0) {
        break;
      }
    default:
      chooseRandom(numOfReplicas, NodeBase.ROOT, excludedNodes, blocksize, maxNodesPerRack, results);
    }
  } catch (NotEnoughReplicasException e) {
    FSNamesystem.LOG.warn("Not able to place enough replicas, still in need of " + numOfReplicas);
  }
  return writer;
}

Recall that we have started with the DFSClient and created a file which was subsequently filled with data. As the blocks need writing out the above code checks first if that can be done on the same host that the client is on, i.e. the "writer". That is "case 0". In "case 1" the code tries to find a remote rack to have a distant replication of the block. Lastly is fills the list of required replicas with local or machines of another rack.

So this means for HBase that as the region server stays up for long enough (which is the default) that after a major compaction on all tables - which can be invoked manually or is triggered by a configuration setting - it has the files local on the same host. The data node that shares the same physical host has a copy of all data the region server requires. If you are running a scan or get or any other use-case you can be sure to get the best performance.

Finally a good overview over the HDFS design and data replication can be found here. Also note that the HBase team is working on redesigning how the Master is assigning the regions to servers. The plan is to improve it so that regions are deployed on the server where most blocks are. This will particularly be useful after a restart because it would guarantee a better data locality right off the bat. Stay tuned!

Minimal Katta Lucene Client

2010-05-14T08:05:00.000-07:00

A quick post explaining how a minimal Katta Lucene Client is set up. I found this was sort of missing from the Katta site and documentation and since I ran into an issue along the way I thought I post my notes here for others who may attempt the same.

First was the question, which of the libs needed to be supplied for a client to use a remote Katta cluster. Please note that I am referring here to a "canonical" setup with a distributed Lucene index (which I created on Hadoop from data in HBase using a MapReduce job). I found these libs needed to be added, the rest is for the server:

katta-core-0.6.rc1.jar
lucene-core-3.0.0.jar
zookeeper-3.2.2.jar
zkclient-0.1-dev.jar
hadoop-core-0.20.1.jar
log4j-1.2.15.jar
commons-logging-1.0.4.jar

Here is the code for the client, please note that this is a simple test app that expects to get the name of the index, the default Lucene search field and query on the command line. I did not add usage info as this is just a proof of concept.

package com.worldlingo.test;

import net.sf.katta.lib.lucene.Hit;
import net.sf.katta.lib.lucene.Hits;
import net.sf.katta.lib.lucene.LuceneClient;
import net.sf.katta.util.ZkConfiguration;
import org.apache.hadoop.io.MapWritable;
import org.apache.hadoop.io.Writable;
import org.apache.lucene.analysis.Analyzer;
import org.apache.lucene.analysis.standard.StandardAnalyzer;
import org.apache.lucene.queryParser.QueryParser;
import org.apache.lucene.search.Query;
import org.apache.lucene.util.Version;

import java.util.Arrays;
import java.util.Map;

public class KattaLuceneClient {

  public static void main(String[] args) {
    try {
      Analyzer analyzer = new StandardAnalyzer(Version.LUCENE_CURRENT);
      Query query = new QueryParser(Version.LUCENE_CURRENT, args[1], analyzer).parse(args[2]);

      // assumes "/katta.zk.properties" available on classpath!
      ZkConfiguration conf = new ZkConfiguration();
      LuceneClient luceneClient = new LuceneClient(conf);
      Hits hits = luceneClient.search(query, Arrays.asList(args[0]).toArray(new String[1]), 99);

      int num = 0;
      for (Hit hit : hits.getHits()) {
        MapWritable mw = luceneClient.getDetails(hit);
        for (Map.Entry<Writable, Writable> entry : mw.entrySet()) {
          System.out.println("[" + (num++) + "] key -> " + entry.getKey() + ", value -> " + entry.getValue());
        }
      }
    } catch (Exception e) {
      e.printStackTrace();
    }
  }

}

The first part is standard Lucene code were we parse the query string with an analyzer. The seconds part is Katta related as it creates a configuration object, which assumes we have a ZooKeeper configuration in the class path. That config only needs to have these lines set:

zookeeper.embedded=false
zookeeper.servers=server-1:2181,server-2:2181

The first line is really only used on the server, so it can be left out on the client. I simply copied the server katta.zk.properties to match my setup. The important line is the second one, which tells the client where the ZooKeeper responsible for managing the Katta cluster is running. With this info the client is able to distribute the search calls to the correct Katta slaves.

Further along we create a LuceneClient instance and start the actual search. Here I simply used no sorting and set the maximum number of hits returned to 99. These two values could be optionally added to the command line parameters but are trivial and not required here - this is a minimal test client after all ;)

The last part of the app is simply printing out the fields and their values of each found document. Please note that Katta is using the low-level Writable class as part of its response. This is not "too" intuitive for the uninitiated. These are actually Text instances so they can safely be convert to text using ".toString()".

Finally, I also checked the test project into my GitHub account for your perusal. Have fun!

3rd Munich OpenHUG Meeting

2010-05-01T10:01:00.000-07:00

I am pleased to invite you to our third Munich Open Hadoop User Group Meeting!

Like always we are looking forward to see everyone again and are welcoming new attendees to join our group. We are enthusiast about all things related to scalable, distributed storage system. We are not limiting us to a particular system but appreciate anyone who would like to share about their experiences.

When: Thursday May 6th, 2010 at 6pm (open end)
Where: eCircle AG, Nymphenburger Straße 86, 80636 München ["Bruckmann" Building, "U1 Mailinger Str", map (in German) and look for the signs]

Thanks again to Bob Schulze from eCircle for providing the infrastructure.

We have a talk scheduled by Stefan Seelmann who is a member of the project committee for the Apache Directory project. This is followed by an open discussion.

Please RSVP at Xing and Yahoo's Upcoming.

Looking forward to seeing you there!

Cheers,
Lars

FOSDEM 2010 NoSQL Talk

2010-02-12T16:01:00.000-08:00

Let me take a minute to wrap up my FOSDEM 2010 experience. I was part of the NoSQL DevRoom organized by @stevenn from Outerthought, who I had the pleasure to visit before a few months back as an HBase Ambassador.

First things first, the NoSQL DevRoom was just an absolute success and I had a blast attending it. I also made sure to not walk around and see other talks outside the NoSQL track while there were many and plenty good ones. I did so deliberately to see the other projects and what they have to offer. I thought it was great, a good vibe was felt throughout the whole day as the audience got a whirlwind tour through the NoSQL landscape. The room was full to the brim for most presentations and some folks had to miss out as we could not have had more enter. This did prove the great interest in this fairly new kind of technology. Exciting!

The focus of my talk was about the history I have with HBase starting with it in late 2007. At this point I would like to take to the opportunity to thank Michael Stack, the lead of HBase, as he has helped me many times back then to sort out the problems I ran into. I would also like to say that if you start with HBase today you will not have these problems as HBase has matured tremendously since then. It is an overall stable product and can solve scalability issues you may face with regular RDBMS's today - and that with great ease.

So the talk I gave did not really sell all the features nor did it explain everything fully. I felt this could be left to the reader to look up on the project's website (or here on my blog) and hence I focused on my use case only. First up, here are the slides.

My Life with HBase - FOSDEM 2010 NoSQL -

After my talk and throughout the rest of the day I also had great conversations with the attendees who had many and great questions.

Having listened to the other talks though I felt I probably could have done a better job selling HBase to the audience. I could have reported about use-cases in well known companies, gave better performance numbers and so on. I have learned a lesson and am making sure I will be doing a better job next time around. I guess this is also another facet of what this is about, i.e. learning to achieve a higher level of professionalism.

But as I said above, my intend was to report about my life with HBase. I am grateful though that it was accepted as that and please let me cite Todd Hoff (see Hot Scalability Links for February 12, 2010) who put it in such nice words:

"The hardscabble tale of HBase's growth from infancy to maturity. A very good introduction and overview of HBase."

Thank you!

Finally here is the video of the talk:

I am looking forward to more NoSQL events in Europe in the near future and will attempt to represent HBase once more (including those adjustments I mentioned above). My hope is that we as Europeans are able to adopt these new technologies and stay abreast with the rest of the world. We sure have smart people to do so.

IvyDE and HBase 0.21 (trunk)

2010-02-05T12:20:00.000-08:00

If you are staying on top of HBase development and frequently update to the HBase trunk (0.21.0-dev at the time of this post) you may have noticed that we now have support for Apache Ivy (see HBASE-1433). This is good because it allows to better control dependencies of the required jar files. It does have a few drawbacks though. One issue that you must be online to get your initial set of jars. You can also set up a local mirror or reuse the one you need for Hadoop anyways to share some of them.

Another issue is that it pulls in many more libs as part of the dependency resolving process. This reminds me bit of aptitude and when you try to install Java, for example on Debian. It often wants to pull in a litany of "required" packages but upon closer look many are only recommended and need not to be installed.

Finally you need to get the jar files somehow into your development environment. I am using Eclipse 3.5 on my Windows 7 PC as well as on my MacOS machines. If you have not run ant from the command line yet you have no jars downloaded and opening the project in Eclipse yields an endless amount of errors. You have two choices, you can run ant and get all jars and then add them to the project in Eclipse. But that is rather static and does not work well with future changes. It also is not the "ivy" way to resolve the libraries automatically.

The other option you have is adding a plugin to Eclipse that can handle Ivy for you, right within the IDE. Luckily for Eclipse there is IvyDE. You install it according to its documentation and then add a "Classpath Container" as described here. That part works quite well and after a restart IvyDE is ready to go.

A few more steps have to be done to get HBase working now - as in compiling without errors. The crucial one is editing the Ivy library and setting the HBase specific Ivy files. In particular the "Ivy settings path" and the properties file. The latter especially is specifying all the various version numbers that the ivy.xml is using. Without it the Ivy resolve process will fail with many errors all over the place. Please note that in the screen shot I added you see how it looks like on my Windows PC. The paths will be slightly different for your setup and probably even using another format if you are on a Mac or Linux machine. As long as you specify both you should be fine.

The other important issue is that you have to repeat that same step adding the Classpath Container two more times: each of the two larger contrib packages "contrib/stargate" and "contrib/transactional" have their own ivy.xml! For both you have to go into the respective directory and right click on the ivy.xml and follow the steps described in the Ivy documentation. Enter the same information as mentioned above to make the resolve work, leave everything else the way it is. You may notice that the contrib packages have a few more targets unticked. That is OK and can be used as-is.

As a temporary step you have to add two more static libraries that are in the $HBASE_HOME/lib directory: libthrift-0.2.0.jar and zookeeper-3.2.2.jar. Those will eventually be published on the Ivy repositories and then this step is obsolete (see INFRA-2461).

Eventually you end up with three containers as shown in the second and third screen shot. The Eclipse toolbar now also has an Ivy "Resolve All Dependencies" button which you can use to trigger the download process. Personally I had to do this a few times as the mirrors with the jars seem to be flaky at times. I ended up with for example "hadoop-mapred.jar" missing. Another resolve run fixed the problem.

The last screen shot shows the three Ivy related containers once more in the tree view of the Package Explorer in the Java perspective. What you also see is the Ivy console, which also is installed with the plugin. You have to open it as usual using the "Window - Show View - Console" menu (if you do not have the Console View open already) and then use the drop down menu next to the "Open Console" button in that view to open the Ivy console. It gives you access to all the details when resolving the dependencies and can hint when you have done something wrong. Please note though that it also lists a lot of connection errors, one for every mirror or repository that does not respond or yet has the required package available. One of them should respond though or as mentioned above you will have to try later again.

Eclipse automatically compiles the project and if everything worked out it does so now without a hitch. Good luck!

Update: Added info about the yet still static thrift and zookeeper jars. See Kay Kay's comment below.

HBase Architecture 101 - Write-ahead-Log

2010-01-30T15:45:00.000-08:00

What is the Write-ahead-Log you ask? In my previous post we had a look at the general storage architecture of HBase. One thing that was mentioned is the Write-ahead-Log, or WAL. This post explains how the log works in detail, but bear in mind that it describes the current version, which is 0.20.3. I will address the various plans to improve the log for 0.21 at the end of this article. For the term itself please read here.

Big Picture

The WAL is the lifeline that is needed when disaster strikes. Similar to a BIN log in MySQL it records all changes to the data. This is important in case something happens to the primary storage. So if the server crashes it can effectively replay that log to get everything up to where the server should have been just before the crash. It also means that if writing the record to the WAL fails the whole operation must be considered a failure.

Let"s look at the high level view of how this is done in HBase. First the client initiates an action that modifies data. This is currently a call to put(Put), delete(Delete) and incrementColumnValue() (abbreviated as "incr" here at times). Each of these modifications is wrapped into a KeyValue object instance and sent over the wire using RPC calls. The calls are (ideally batched) to the HRegionServer that serves the affected regions. Once it arrives the payload, the said KeyValue, is routed to the HRegion that is responsible for the affected row. The data is written to the WAL and then put into the MemStore of the actual Store that holds the record. And that also pretty much describes the write-path of HBase.

Eventually when the MemStore gets to a certain size or after a specific time the data is asynchronously persisted to the file system. In between that timeframe data is stored volatile in memory. And if the HRegionServer hosting that memory crashes the data is lost... but for the existence of what is the topic of this post, the WAL!

We have a look now at the various classes or "wheels" working the magic of the WAL. First up is one of the main classes of this contraption.

HLog

The class which implements the WAL is called HLog. What you may have read in my previous post and is also illustrated above is that there is only one instance of the HLog class, which is one per HRegionServer. When a HRegion is instantiated the single HLog is passed on as a parameter to the constructor of HRegion.

Central part to HLog's functionality is the append() method, which internally eventually calls doWrite(). It is what is called when the above mentioned modification methods are invoked... or is it? One thing to note here is that for performance reasons there is an option for put(), delete(), and incrementColumnValue() to be called with an extra parameter set: setWriteToWAL(boolean). If you invoke this method while setting up for example a Put() instance then the writing to WAL is forfeited! That is also why the downward arrow in the big picture above is done with a dashed line to indicate the optional step. By default you certainly want the WAL, no doubt about that. But say you run a large bulk import MapReduce job that you can rerun at any time. You gain extra performance but need to take extra care that no data was lost during the import. The choice is yours.

Another important feature of the HLog is keeping track of the changes. This is done by using a "sequence number". It uses an AtomicLong internally to be thread-safe and is either starting out at zero - or at that last known number persisted to the file system. So as the region is opening its storage file, it reads the highest sequence number which is stored as a meta field in each HFile and sets the HLog sequence number to that value if it is higher than what has been recorded before. So at the end of opening all storage files the HLog is initialized to reflect where persisting has ended and where to continue. You will see in a minute where this is used.

The image to the right shows three different regions. Each of them covering a different row key range. As mentioned above each of these regions shares the the same single instance of HLog. What that means in this context is that the data as it arrives at each region it is written to the WAL in an unpredictable order. We will address this further below.

Finally the HLog has the facilities to recover and split a log left by a crashed HRegionServer. These are invoked by the HMaster before regions are deployed again.

HLogKey

Currently the WAL is using a Hadoop SequenceFile, which stores record as sets of key/values. For the WAL the value is simply the KeyValue sent from the client. The key is represented by an HLogKey instance. If you may recall from my first post in this series the KeyValue does only represent the row, column family, qualifier, timestamp, and value as well as the "Key Type". Last time I did not address that field since there was no context. Now we have one because the Key Type is what identifies what the KeyValue represents, a "put" or a "delete" (where there are a few more variations of the latter to express what is to be deleted, value, column family or a specific column).

What we are missing though is where the KeyValue belongs to, i.e. the region and the table name. That is stored in the HLogKey. What is also stored is the above sequence number. With each record that number is incremented to be able to keep a sequential order of edits. Finally it records the "Write Time", a time stamp to record when the edit was written to the log.

LogFlusher

As mentioned above as data arrives at a HRegionServer in form of KeyValue instances it is written (optionally) to the WAL. And as mentioned as well it is then written to a SequenceFile. While this seems trivial, it is not. One of the base classes in Java IO is the Stream. Especially streams writing to a file system are often buffered to improve performance as the OS is much faster writing data in batches, or blocks. If you write records separately IO throughput would be really bad. But in the context of the WAL this is causing a gap where data is supposedly written to disk but in reality it is in limbo. To mitigate the issue the underlaying stream needs to be flushed on a regular basis. This functionality is provided by the LogFlusher class and thread. It simply calls HLog.optionalSync(), which checks if the hbase.regionserver.optionallogflushinterval, set to 10 seconds by default, has been exceeded and if that is the case invokes HLog.sync(). The other place invoking the sync method is HLog.doWrite(). Once it has written the current edit to the stream it checks if the hbase.regionserver.flushlogentries parameter, set to 100 by default, has been exceeded and calls sync as well.

Sync itself invokes HLog.Writer.sync() and is implemented in SequenceFileLogWriter. For now we assume it flushes the stream to disk and all is well. That in reality this is all a bit more complicated is discussed below.

LogRoller

Obviously it makes sense to have some size restrictions related to the logs written. Also we want to make sure a log is persisted on a regular basis. This is done by the LogRoller class and thread. It is controlled by the hbase.regionserver.logroll.period parameter in the $HBASE_HOME/conf/hbase-site.xml file. By default this is set to 1 hour. So every 60 minutes the log is closed and a new one started. Over time we are gathering that way a bunch of log files that need to be maintained as well. The HLog.rollWriter() method, which is called by the LogRoller to do the above rolling of the current log file, is taking care of that as well by calling HLog.cleanOldLogs() subsequently. It checks what the highest sequence number written to a storage file is, because up to that number all edits are persisted. It then checks if there is a log left that has edits all less than that number. If that is the case it deletes said logs and leaves just those that are still needed.

This is a good place to talk about the following obscure message you may see in your logs:

2009-12-15 01:45:48,427 INFO org.apache.hadoop.hbase.regionserver.HLog: Too

many hlogs: logs=130, maxlogs=96; forcing flush of region with oldest edits:

foobar,1b2dc5f3b5d4,1260083783909

It is printed because the configured maximum number of log files to keep exceeds the number of log files that are required to be kept because they still contain outstanding edits that have not yet been persisted. The main reason I saw this being the case is when you stress out the file system so much that it cannot keep up persisting the data at the rate new data is added. Otherwise log flushes should take care of this. Note though that when this message is printed the server goes into a special mode trying to force flushing out edits to reduce the number of logs required to be kept.

The other parameters controlling the log rolling are hbase.regionserver.hlog.blocksize and hbase.regionserver.logroll.multiplier, which are set by default to rotate logs when they are at 95% of the blocksize of the SequenceFile, typically 64M. So either the logs are considered full or when a certain amount of time has passed causes the logs to be switched out, whatever comes first.

Replay

Once a HRegionServer starts and is opening the regions it hosts it checks if there are some left over log files and applies those all the way down in Store.doReconstructionLog(). Replaying a log is simply done by reading the log and adding the contained edits to the current MemStore. At the end an explicit flush of the MemStore (note, this is not the flush of the log!) helps writing those changes out to disk.

The old logs usually come from a previous region server crash. When the HMaster is started or detects that region server has crashed it splits the log files belonging to that server into separate files and stores those in the region directories on the file system they belong to. After that the above mechanism takes care of replaying the logs. One thing to note is that regions from a crashed server can only be redeployed if the logs have been split and copied. Splitting itself is done in HLog.splitLog(). The old log is read into memory in the main thread (means single threaded) and then using a pool of threads written to all region directories, one thread for each region.

Issues

As mentioned above all edits are written to one HLog per HRegionServer. You would ask why that is the case? Why not write all edits for a specific region into its own log file? Let's quote the BigTable paper once more:

If we kept the commit log for each tablet in a separate log file, a very large number of files would be written concurrently in GFS. Depending on the underlying file system implementation on each GFS server, these writes could cause a large number of disk seeks to write to the different physical log files.

HBase followed that principle for pretty much the same reasons. As explained above you end up with many files since logs are rolled and kept until they are safe to be deleted. If you do this for every region separately this would not scale well - or at least be an itch that sooner or later is causing pain.

So far that seems to be no issue. But again, it causes problems when things go wrong. As long as you have applied all edits in time and persisted the data safely, all is well. But if you have to split the log because of a server crash then you need to divide into suitable pieces, as described above in the "replay" paragraph. But as you have seen above as well all edits are intermingled in the log and there is no index of what is stored at all. For that reason the HMaster cannot redeploy any region from a crashed server until it has split the logs for that very server. And that can be quite a number if the server was behind applying the edits.

Another problem is data safety. You want to be able to rely on the system to save all your data, no matter what newfangled algorithms are employed behind the scenes. As far as HBase and the log is concerned you can turn down the log flush times to as low as you want - you are still dependent on the underlaying file system as mentioned above; the stream used to store the data is flushed but is it written to disk yet? We are talking about fsync style issues. Now for HBase we are most likely talking Hadoop's HDFS as being the file system that is persisted to.

Up to this point it should be abundantly clear that the log is what keeps data safe. For that reason a log could be kept open for up to an hour (or more if configured so). As data arrives a new key/value pair is written to the SequenceFile and occasionally flushed to disk. But that is not how Hadoop was set out to work. It was meant to provide an API that allows to open a file, write data into it (preferably a lot) and closed right away, leaving an immutable file for everyone else to read many times. Only after a file is closed it is visible and readable to others. If a process dies while writing the data the file is pretty much considered lost. What is required is a feature that allows to read the log up to the point where the crashed server has written it (or as close as possible).

Interlude: HDFS append, hflush, hsync, sync... wth?

It all started with HADOOP-1700 reported by HBase lead Michael Stack. It was committed in Hadoop 0.19.0 and meant to solve the problem. But that was not the case. So the issue was tackled again in HADOOP-4379 aka HDFS-200 and implemented syncFs() that was meant to help syncing changes to a file to be more reliable. For a while we had custom code (see HBASE-1470) that detected a patched Hadoop that exposed that API. But again this did not solve the issue entirely.

Then came HDFS-265, which revisits the append idea in general. It also introduces a Syncable interface that exposes hsync() and hflush().

Lastly SequenceFile.Writer.sync() is not the same as the above, it simply writes a synchronization marker into the file that helps reading it later - or recover data if broken.

While append for HDFS in general is useful it is not used in HBase, but the hflush() is. What it does is writing out everything to disk as the log is written. In case of a server crash we can safely read that "dirty" file up to the last edits. The append in Hadoop 0.19.0 was so badly suited that a hadoop fsck / would report the DFS being corrupt because of the open log files HBase kept.

Bottom line is, without Hadoop 0.21.0 you can very well face data loss. With Hadoop 0.21.0 you have a state-of-the-art system.

Planned Improvements

For HBase 0.21.0 there are quite a few things lined up that affect the WAL architecture. Here are some of the noteworthy ones.

SequenceFile Replacement

One of the central building blocks around the WAL is the actual storage file format. The used SequenceFile has quite a few shortcomings that need to be addressed. One for example is the suboptimal performance as all writing in SequenceFile is synchronized, as documented in HBASE-2105.

As with HFile replacing MapFile in HBase 0.20.0 it makes sense to think about a complete replacement. A first step was done to make the HBase classes independent of the underlaying file format. HBASE-2059 made the class implementing the log configurable.

Another idea is to change to a different serialization altogether. HBASE-2055 proposes such a format using Hadoop's Avro as the low level system. Avro is also slated to be the new RPC format for Hadoop, which does help as more people are familiar with it.

Append/Sync

Even with hflush() we have a problem that calling it too often may cause the system to slow down. Previous tests using the older syncFs() call did show that calling it for every record slows down the system considerably. One step to help is to implement a "Group Commit", done in HBASE-1939. It flushes out records in batches. In addition HBASE-1944 adds the notion of a "deferred log flush" as a parameter of a Column Family. If set to true it leaves the syncing of changes to the log to the newly added LogSyncer class and thread. Finally HBASE-2041 sets the flushlogentries to 1 and optionallogflushinterval to 1000 msecs. The .META. is always synced for every change, user tables can be configured as needed.

Distributed Log Splitting

As remarked splitting the log is an issue when regions need to be redeployed. One idea is to keep a list of regions with edits in Zookeeper. That way at least all "clean" regions can be deployed instantly. Only those with edits need to wait then until the logs are split.

What is left is to improve how the logs are split to make the process faster. Here is how is the BigTable addresses the issue:

One approach would be for each new tablet server to read this full commit log file and apply just the entries needed for the tablets it needs to recover. However, under such a scheme, if 100 machines were each assigned a single tablet from a failed tablet server, then the log file would be read 100 times (once by each server).

and further

We avoid duplicating log reads by first sorting the commit log entries in order of the keys (table, row name, log sequence number). In the sorted output, all mutations for a particular tablet are contiguous and can therefore be read efficiently with one disk seek followed by a sequential read. To parallelize the sorting, we partition the log file into 64 MB segments, and sort each segment in parallel on different tablet servers. This sorting process is coordinated by the master and is initiated when a tablet server indicates that it needs to recover mutations from some commit log file.

This is where its at. As part of the HMaster rewrite (see HBASE-1816) the log splitting will be addressed as well. HBASE-1364 wraps the splitting of logs into one issue. But I am sure that will evolve in more sub tasks as the details get discussed.

2nd Munich OpenHUG Meeting

2010-01-28T04:47:00.000-08:00

At the end of last year we had a first meeting in Munich to allow everyone interested in all things Hadoop and related technologies to gather, get to know each other and exchange their knowledge, experiences and information. We would like to invite for the second Munich OpenHUG Meeting and hope to see you all again and meet even more new enthusiasts there. We would also be thrilled if those attending would be willing to share their story so that we can learn about other projects and how people are using the exciting NoSQL related technologies. No pressure though, come along and simply listen if you prefer, we welcome anyone and everybody!

When: Thursday February 25, 2010 at 5:30pm open end
Where: eCircle AG, Nymphenburger Straße 86, 80636 München ["Bruckmann" Building, "U1 Mailinger Str", map (in German) and look for the signs]

Thanks again to Bob Schulze from eCircle to provide the location and projector. So far we have a talk scheduled by Christoph Rupp about HamsterDB. We are still looking for volunteers who would like to present on any related topic (please contact me)! Otherwise we will have an open discussion about whatever is brought up by the attendees.

Last but not least there will be something to drink and we will get pizzas in. Since we do not know how many of you will come we simply stay at the events location and continue our chats over food.

Looking forward to seeing you there!

Please RSVP at Upcoming or Xing.

First Munich OpenHUG Meeting - Summary

2010-01-10T11:52:00.000-08:00

On December 17th we had the first Munich OpenHUG meeting. The location was kindly provided by eCircle's Bob Schulze. This was the first in a series of meetings in Munich based on everything Hadoop and related technologies. Let's use the term NoSQL with care as this is not about blame or finger pointing (I feel that it was used like that in the past and therefore making this express point). We are trying to get together the brightest local and remote talent to report and present on new age and evolved existing technologies.

The first talk of the night was given by Bob himself presenting his findings of evaluating HBase and Hadoop for their internal use. He went into detail explaining how HBase is structuring its data and how it can be used for their needs. One thing that I noted in particular was his HBase Explorer, which he subsequently published on SourceForge as an Open Source project. The talk was concluded by an open discussion about HBase.

The second part of the meeting was my own presentation about how we at WorldLingo use HBase and Hadoop (as well as Lucene etc.)

We continued our discussion on HBase with the developers of eCircle present. This was very interesting and fruitful and we had the chance to exchange experiences made along our similar paths.

I would have wished for the overall attendance to be a little higher, but it was a great start. Talking to other hosts of similar events it seems that this is normal and therefore my hopes are up for the next meetings throughout this year. We have planned the next meeting for February 25th, 2010 at the same location. If you have interest in presenting a talk on any related topic, please contact me!

I am looking forward to meeting you all there!

Lars

First Munich OpenHUG Meeting

2009-12-14T01:22:00.000-08:00

First Munich OpenHUG Meeting

We are trying to gauge the interest in a south Germany Hadoop User Group Meeting. After seeing quite a big interest in the Berlin meetings a few of us got together and decided to test the waters for another meeting at the other end of the country. We are therefore happy to announce the first Munich OpenHUG Meeting.

When: Thursday December 17, 2009 at 5:30pm open end
Where: eCircle AG, Nymphenburger Straße 86, 80636 München ("Bruckmann" Building, "U1 Mailinger Str", map in German and look for the signs)

Thanks to Bob Schulze from eCircle to provide the location, projector and also giving a first presentation on how eCircle is planning to use the Hadoop stack.

We also have Dave Butlerdi giving an overview of his usage of Hadoop.
Finally I will give a state of affairs of the HBase project. What is it, what does it do and how am I using it (since early 2008).

We are also open for everyone who wants to talk about anything related to these new technologies often combined under the rather new term "NoSQL". Take the opportunity to talk about what you are working on and find like minded people to bounce ideas off. This is also why we chose the title OpenHUG for the meeting. While we mostly work with Hadoop and its subprojects we also like to learn about related projects and technologies.

Last but not least there will be something to drink and we will get pizzas in. Since we do not know how many of you will come on such short notice we simply stay at Bob's place and continue or chats over food.

As this is a first meeting in Munich on this topic we called it in a day after the Berlin meeting. Given there is interest we will in the future settle on dates that fit nicely between the Berlin dates so that we have no overlap and you can attend both meetings.

Please RSVP at Yahoo's Upcoming or Xing.

HBase vs. BigTable Comparison

2009-11-24T06:08:00.000-08:00

HBase is an open-source implementation of the Google BigTable architecture. That part is fairly easy to understand and grasp. What I personally feel is a bit more difficult is to understand how much HBase covers and where there are differences (still) compared to the BigTable specification. This post is an attempt to compare the two systems.

Before we embark onto the ~~dark~~ technology side of things I would like to point out one thing upfront: HBase is very close to what the BigTable paper describes. Putting aside minor differences, as of HBase 0.20, which is using ZooKeeper as its ~~lock~~ distributed coordination service, it has all the means to be nearly an exact implementation of BigTable's functionality. What I will be looking into below are mainly subtle variations or differences. Where possible I will try to point out how the HBase team is working on improving the situation given there is a need to do so.

Scope

The comparison in this post is based on the OSDI'06 paper that describes the system Google implemented in about seven person-years and which is in operation since 2005. The paper was published 2006 while the HBase sub-project of Hadoop was established only around the end of that same year to early 2007. Back then the current version of Hadoop was 0.15.0. Given we are now about 2 years in, with Hadoop 0.20.1 and HBase 0.20.2 available, you can hopefully understand that indeed much has happened since. Please also note that I am comparing a 14 page high level technical paper with an open-source project that can be examined freely from top to bottom. It usually means that there is more to tell about how HBase does things because the information is available.

Towards the end I will also address a few newer features that BigTable has nowadays and how HBase is comparing to those. We start though with naming conventions.

Terminology

There are a few different terms used in either system describing the same thing. The most prominent being what HBase calls "regions" while Google refers to it as "tablet". These are the partitions of subsequent rows spread across many "region servers" - or "tablet server" respectively. Apart from that most differences are minor or caused by usage of related technologies since Google's code is obviously closed-source and therefore only mirrored by open-source projects. The open-source projects are free to use other terms and most importantly names for the projects themselves.

Features

The following table lists various "features" of BigTable and compares them with what HBase has to offer. Some are actual implementation details, some are configurable option and so on. This may be confusing but it would be difficult to sort them into categories and not ending up with one entry only in each of them.

Feature	Google BigTable	Apache HBase	Notes
Atomic Read/Write/Modify	Yes, per row	Yes, per row	Since BigTable does not strive to be a relational database it does not have transactions. The closest to such a mechanism is the atomic access to each row in the table. HBase also implements a row lock API which allows the user to lock more than one row at a time.
Lexicographic Row Order	Yes	Yes	All rows are sorted lexicographically in one order and that one order only. Again, this is no SQL database where you can have different sorting orders.
Block Support	Yes	Yes	Within each storage file data is written as smaller blocks of data. This enables faster loading of data from large storage files. The size is configurable in either system. The typical size is 64K.
Block Compression	Yes, per column family	Yes, per column family	Google uses BMDiff and Zippy in a two step process. BMDiff works really well because neighboring key-value pairs in the store files are often very similar. This can be achieved by using versioning so that all modifications to a value are stored next to each other but still have a lot in common. Or by designing the row keys in such a way that for example web pages from the same site are all bundled. Zippy then is a modified LZW algorithm. HBase on the other hand uses the standard Java supplied GZip or with a little effort the GPL licensed LZO format. There are indications though that Hadoop also may want to have BMDiff (HADOOP-5793) and possibly Zippy as well.
Number of Column Families	Hundreds at Most	Less than 100	While the number of rows and columns is theoretically unbound the number of column families is not. This is a design trade-off but does not impose too much restrictions if the tables and key are designed accordingly.
Column Family Name Format	Printable	Printable	The main reason for HBase here is that column family names are used as directories in the file system.
Qualifier Format	Arbitrary	Arbitrary	Any arbitrary byte[] array can be used.
Key/Value Format	Arbitrary	Arbitrary	Like above, any arbitrary byte[] array can be used.
Access Control	Yes	No	BigTable enforces access control on a column family level. HBase does not have yet have that feature (see HBASE-1697).
Cell Versions	Yes	Yes	Versioning is done using timestamps. See next feature below too. The number of versions that should be kept are freely configurable on a column family level.
Custom Timestamps	Yes (micro)	Yes (milli)	With both systems you can either set the timestamp of a value that is stored yourself or leave the default "now". There are "known" restrictions in HBase that the outcome is indeterminate when adding older timestamps after already having stored newer ones beforehand.
Data Time-To-Live	Yes	Yes	Besides having versions of data cells the user can also set a time-to-live on the stored data that allows to discard data after a specific amount of time.
Batch Writes	Yes	Yes	Both systems allow to batch table operations.
Value based Counters	Yes	Yes	BigTable and HBase can use a specific column as atomic counters. HBase does this by acquiring a row lock before the value is incremented.
Row Filters	Yes	Yes	Again both system allow to apply filters when scanning rows.
Client Script Execution	Yes	No	BigTable uses Sawzall to enable users to process the stored data.
MapReduce Support	Yes	Yes	Both systems have convenience classes that allow scanning a table in MapReduce jobs.
Storage Systems	GFS	HDFS, S3, S3N, EBS	While BigTable works on Google's GFS, HBase has the option to use any file system as long as there is a proxy or driver class for it.
File Format	SSTable	HFile
Block Index	At end of file	At end of file	Both storage file formats have a similar block oriented structure with the block index stored at the end of the file.
Memory Mapping	Yes	No	BigTable can memory map storage files directly into memory.
Lock Service	Chubby	ZooKeeper	There is a difference in where ZooKeeper is used to coordinate tasks in HBase as opposed to provide locking services. Overall though ZooKeeper does for HBase pretty much what Chubby does for BigTable with slightly different semantics.
Single Master	Yes	No	HBase recently added support for multiple masters. These are on "hot" standby and monitor the master's ZooKeeper node.
Tablet/Region Count	10-1000	10-1000	Both systems recommend about the same amount of regions per region server. Of course this depends on many things but given a similar setup as far as "commodity" machines are concerned it seems to result in the same amount of load on each server.
Tablet/Region Size	100-200MB	256MB	The maximum region size can be configured for HBase and BigTable. HBase used 256MB as the default value.
Root Location	1st META / Chubby	-ROOT- / ZooKeeper	HBase handles the Root table slightly different from BigTable, where it is the first region in the Meta table. HBase uses its own table with a single region to store the Root table. Once either system starts the address of the server hosting the Root region is stored in ZooKeeper or Chubby so that the clients can resolve its location without hitting the master.
Client Region Cache	Yes	Yes	The clients in either system caches the location of regions and has appropriate mechanisms to detect stale information and update the local cache respectively
Meta Prefetch	Yes	No (?)	A design feature of BigTable is to fetch more than one Meta region information. This proactively fills the client cache for future lookups.
Historian	Yes	Yes	The history of region related events (such as splits, assignment, reassignment) is recorded in the Meta table.
Locality Groups	Yes	No	It is not entirely clear but it seems everything in BigTable is defined by Locality Groups. The group multiple column families into one so that they get stored together and also share the same configuration parameters. A single column family is probably a Locality Group with one member. HBase does not have this option and handles each column family separately.
In-Memory Column Families	Yes	Yes	These are for relatively small tables that need very fast access times.
KeyValue (Cell) Cache	Yes	No	This is a cache that servers hot cells.
Block Cache	Yes	Yes	Blocks read from the storage files are cached internally in configurable caches.
Bloom Filters	Yes	Yes	These filters allow - at a cost of using memory on the region server - to quickly check if a specific cell exists or maybe not.
Write-Ahead Log (WAL)	Yes	Yes	Each region server in either system stores one modification log for all regions it hosts.
Secondary Log	Yes	No	In addition to the Write-Ahead log mentioned above BigTable has a second log that it can use when the first is going slow. This is a performance optimization.
Skip Write-Ahead Log	?	Yes	For bulk imports the client in HBase can opt to skip writing into the WAL.
Fast Table/Region Split	Yes	Yes	Splitting a region or tablet is fast as the daughter regions first read the original storage file until a compaction finally rewrites the data into the region's local store.

New Features

As mentioned above, a few years have passed since the original OSDI'06 BigTable paper. Jeff Dean - a fellow at Google - has mentioned a few new BigTable features during speeches and presentations he gave recently. We will have a look at some of them here.

Feature	Google BigTable	Apache HBase	Notes
Client Isolation	Yes	No	BigTable is internally used to server many separate clients and can therefore keep the data between isolated.
Coprocessors	Yes	No	BigTable can host code that resides with the regions and splits with them as well. See HBASE-2000 for progress on this feature within HBase.
Corruption Safety	Yes	No	This is an interesting topic. BigTable uses CRC checksums to verify if data has been written safely. While HBase does not have this, the question is if that is build into Hadoop's HDFS?
Replication	Yes	No	HBase is working on the same topic in HBASE-1295

Note: the color codes indicate what features have a direct match or where it is missing (yet). Weaker features are colored yellow, as I am not sure if they are immediately necessary or even applicable given HBase's implementation.

Variations and Differences

Some of the above features need a bit more looking into as they are difficult to be narrowed down to simple "Yay or Nay" questions. I am addressing them below separately.

Lock Service

This is from the BigTable paper:

Bigtable uses Chubby for a variety of tasks: to ensure that there is at most one active master at any time; to store the bootstrap location of Bigtable data (see Section 5.1); to discover tablet servers and finalize tablet server deaths (see Section 5.2); to store Bigtable schema information (the column family information for each table); and to store access control lists. If Chubby becomes unavailable for an extended period of time, Bigtable becomes unavailable.

There is a lot of overlap compared to how HBase does use ZooKeeper. What is different though is that schema information is not stored in ZooKeeper (yet, see http://wiki.apache.org/hadoop/ZooKeeper/HBaseUseCases) for details. What is important here though is the same reliance on the lock service being available. From my own experience and reading the threads on the HBase mailing list it is often underestimated what can happen when ZooKeeper does not get the resources it needs to react timely. It is better to have a small ZooKeeper cluster on older machines not doing anything else as opposed to having ZooKeeper nodes running next to the already heavy Hadoop or HBase processes. Once you starve ZooKeeper you will see a domino effect of HBase nodes going down with it - including the master(s).

Update: After talking to a few guys of the ZooKeeper team I would like to point out that this is indeed not a ZooKeeper issue. It has to do with the fact that if you have an already heavily loaded node trying to also respond in time to ZooKeeper resources then you may face a timeout situation where the HBase RegionServers and even the Master may think that their coordination service is gone and shut themselves down. Patrick Hunt has responded to this by mail and by post. Please read both to see that ZooKeeper is able to handle the load. I personally recommend to set up ZooKeeper in combination with HBase on a separate cluster, maybe a set of spare machines you have from a recent update to the cluster and which are slightly outdated (no 2xquad core CPU with 16GB of memory) but are otherwise perfectly fine. This also allows you to monitor the machines separately and not having to see a combined CPU load of 100% on the servers and not really knowing where it comes from and what effect it may have.

Another important difference is that ZooKeeper is no lock service like Chubby - and I do think it does not have to be as far as HBase is concerned. ZooKeeper is a distributed coordination service enabling HBase to do Master node elections etc. It also allows using semaphores to indicate state or actions required. So where Chubby creates a lock file to indicate a tablet server is up and running HBase in turn uses ephemeral nodes that exist as long as the session between the RegionServer which creates that node and ZooKeeper is active. This also causes the differences in semantics where in BigTable can delete a tablet servers lock file to indicate that it has lost its lease on tablets. In HBase this has to be handled differently because of the slightly less restrictive architecture of ZooKeeper. These are only semantics as mentioned and do not mean one is better than the other - just different.

The first level is a file stored in Chubby that contains the location of the root tablet. The root tablet contains the location of all tablets in a special METADATA table. Each METADATA tablet contains the location of a set of user tablets. The root tablet is just the first tablet in the METADATA table, but is treated specially - it is never split - to ensure that the tablet location hierarchy has no more than three levels.

As mentioned above in HBase the root region is its own table with a single region. If that makes a difference to having it as the first (non-splittable) region of the meta table I doubt strongly. It is just the same feature but implemented differently.

The METADATA table stores the location of a tablet under a row key that is an encoding of the tablet's table identifier and its end row.

HBase does have a different layout here. It stores the start and end row with each region where the end row is exclusive and denotes the first (or start) row of the next region. Again, these are minor differences and I am not sure if there is a better or worse solution. It is just done differently.

Master Operation

To detect when a tablet server is no longer serving its tablets, the master periodically asks each tablet server for the status of its lock. If a tablet server reports that it has lost its lock, or if the master was unable to reach a server during its last several attempts, the master attempts to acquire an exclusive lock on the server's file. If the master is able to acquire the lock, then Chubby is live and the tablet server is either dead or having trouble reaching Chubby, so the master ensures that the tablet server can never serve again by deleting its server file. Once a server's file has been deleted, the master can move all the tablets that were previously assigned to that server into the set of unassigned tablets. To ensure that a Bigtable cluster is not vulnerable to networking issues between the master and Chubby, the master kills itself if its Chubby session expires.

This is quite different even up to the current HBase 0.20.2. Here the master uses a heartbeat protocol that is used by the region servers to report for duty and that they are still alive subsequently. I am not sure if this topic is covered by the master rewrite umbrella issue HBASE-1816 - and if it needs to be addressed at all. It could well be that what we have in HBase now is sufficient and does its job just fine. It was created when there was no lock service yet and therefore could be considered legacy code too.

Master Startup

The master executes the following steps at startup. (1) The master grabs a unique master lock in Chubby, which prevents concurrent master instantiations. (2) The master scans the servers directory in Chubby to find the live servers. (3) The master communicates with every live tablet server to discover what tablets are already assigned to each server. (4) The master scans the METADATA table to learn the set of tablets. Whenever this scan encounters a tablet that is not already assigned, the master adds the tablet to the set of unassigned tablets, which makes the tablet eligible for tablet assignment.

Along what I mentioned above, this part of the code was created before ZooKeeper was available. So HBase actually waits for the region servers to report for duty. It also scans the .META. table to learn what is there and which server is assigned to it. ZooKeeper is (yet) only used to publish the server hosting the -ROOT- region.

Tablet/Region Splits

In case the split notification is lost (either because the tablet server or the master died), the master detects the new tablet when it asks a tablet server to load the tablet that has now split. The tablet server will notify the master of the split, because the tablet entry it finds in the METADATA table will specify only a portion of the tablet that the master asked it to load.

The master node in HBase uses the .META. solely to detect when a region was split but the message was lost. For that reason it scans the .META. on a regular basis to see when a region appears that is not yet assigned. It will then assign that region as per its default strategy.

Compactions

The following are more terminology differences than anything else.

As write operations execute, the size of the memtable increases. When the memtable size reaches a threshold, the memtable is frozen, a new memtable is created, and the frozen memtable is converted to an SSTable and written to GFS. This minor compaction process has two goals: it shrinks the memory usage of the tablet server, and it reduces the amount of data that has to be read from the commit log during recovery if this server dies. Incoming read and write operations can continue while compactions occur.

HBase has a similar operation but it is referred to as a "flush". Opposed to that "minor compactions" in HBase rewrite the last N used store files, i.e. those with the most recent mutations as they are probably much smaller than previously created files that have more data in them.

... we bound the number of such files by periodically executing a merging compaction in the background. A merging compaction reads the contents of a few SSTables and the memtable, and writes out a new SSTable. The input SSTables and memtable can be discarded as soon as the compaction has finished.

This again refers to what is called "minor compaction" in HBase.

A merging compaction that rewrites all SSTables into exactly one SSTable is called a major compaction.

Here we have an exact match though, a "major compaction" in HBase also rewrites all files into one.

Immutable Files

Knowing that files are fixed once written BigTable makes the following assumption:

The only mutable data structure that is accessed by both reads and writes is the memtable. To reduce contention during reads of the memtable, we make each memtable row copy-on-write and allow reads and writes to proceed in parallel.

I do believe this is done similar in HBase but am not sure. It certainly has the same architecture as HDFS files for example are also immutable once written.

I can only recommend that you read the BigTable too and make up your own mind. This post was inspired by the idea to learn what BigTable really has to offer and how much HBase has already covered. The difficult part is of course that there is not too much information available on BigTable. But the numbers even the 2006 paper lists are more than impressive. If HBase as on open-source project with just a handful of committers of whom most have a full-time day jobs can achieve something even remotely comparable I think this is a huge success. And looking at the 0.21 and 0.22 road map, the already small gap is going to shrink even further!

HBase on Cloudera Training Virtual Machine (0.3.2)

2009-11-20T06:05:00.000-08:00

Note: This is a follow up to my earlier post. Since then Cloudera released a new VM that includes the current 0.20 branch of Hadoop. Below I have the same post adjusted to work with that new release. Please note that there are subtle changes, for example the NameNode port has changed. So if you in any way still have the older post please make sure you forget about it and follow this one here instead for the new VM version.

You might want to run HBase on Cloudera's Virtual Machine to get a quick start to a prototyping setup. In theory you download the VM, start it and you are ready to go. The main issue though is that the current Hadoop Training VM does not include HBase at all (yet?). Apart from that the install of a local HBase instance is a straight forward process.

Here are the steps to get HBase running on Cloudera's VM:

Download VM

Get it from Cloudera's website.
Start VM

As the above page states: "To launch the VMWare image, you will either need VMware Player for windows and linux, or VMware Fusion for Mac."

Note: I have Parallels for Mac and wanted to use that. I used Parallels Transporter to convert the "cloudera-training-0.3.2.vmx" to a new "cloudera-training-0.2-cl4-000001.hdd", create a new VM in Parallels selecting Ubuntu Linux as the OS and the newly created .hdd as the disk image. Boot up the VM and you are up and running. I gave it a bit more memory for the graphics to be able to switch the VM to 1440x900 which is the native screen resolution on my MacBook Pro I am using.

Finally follow the steps explained on the page above, i.e. open a Terminal and issue:
```
$ cd ~/git
$ ./update-exercises --workspace
```
Pull HBase branch

We are using the brand new HBase 0.20.2 release. Open a new Terminal (or issue a $ cd .. in the open one), then:
```
$ sudo -u hadoop git clone http://git.apache.org/hbase.git /home/hadoop/hbase
$ sudo -u hadoop sh -c "cd /home/hadoop/hbase ; git checkout origin/tags/0.20.2"
Note: moving to "origin/tags/0.20.2" which isn't a local branch
If you want to create a new branch from this checkout, you may do so
(now or later) by using -b with the checkout command again. Example:
  git checkout -b <new_branch_name>
HEAD is now at 777fb63... HBase release 0.20.2
```
First we clone the repository, then switch to the actual branch. You will notice that I am using sudo -u hadoop because Hadoop itself is started under that account and so I wanted it to match. Also, the default "training" account does not have SSH set up as explained in Hadoop's quick-start guide. When sudo is asking for a password use the default, which is set to "training".

You can ignore the messages git prints out while performing the checkout.

Build Branch

Continue in Terminal:

$ sudo -u hadoop sh -c "cd /home/hadoop/hbase/ ; export PATH=$PATH:/usr/share/apache-ant-1.7.1/bin ; ant package"
...
BUILD SUCCESSFUL

Configure HBase

There are a few edits to be made to get HBase running.

$ sudo -u hadoop vim /home/hadoop/hbase/build/conf/hbase-site.xml

<configuration>

  <property>
    <name>hbase.rootdir</name>
    <value>hdfs://localhost:8022/hbase</value>
  </property>

</configuration>

$ sudo -u hadoop vim /home/hadoop/hbase/build/conf/hbase-env.sh 

# The java implementation to use.  Java 1.6 required.
# export JAVA_HOME=/usr/java/jdk1.6.0/
export JAVA_HOME=/usr/lib/jvm/java-6-sun
...

Rev up the Engine!

The final thing is to start HBase:

$ sudo -u hadoop /home/hadoop/hbase/build/bin/start-hbase.sh

$ sudo -u hadoop /home/hadoop/hbase/build/bin/hbase shell
HBase Shell; enter 'help<RETURN>' for list of supported commands.
Version: 0.20.2, r777fb63ff0c73369abc4d799388a45b8bda9e5fd, Thu Nov 19 15:32:17 PST 2009
hbase(main):001:0>

Done!

Let's create a table and check if it was created OK.

hbase(main):001:0> list
0 row(s) in 0.0910 seconds

hbase(main):002:0> create 't1', 'f1', 'f2', 'f3'
0 row(s) in 6.1260 seconds

hbase(main):003:0> list                         
t1                                                                                                            
1 row(s) in 0.0470 seconds

hbase(main):004:0> describe 't1'                
DESCRIPTION                                                             ENABLED                               
 {NAME => 't1', FAMILIES => [{NAME => 'f1', COMPRESSION => 'NONE', VERS true                                  
 IONS => '3', TTL => '2147483647', BLOCKSIZE => '65536', IN_MEMORY => '                                       
 false', BLOCKCACHE => 'true'}, {NAME => 'f2', COMPRESSION => 'NONE', V                                       
 ERSIONS => '3', TTL => '2147483647', BLOCKSIZE => '65536', IN_MEMORY =                                       
 > 'false', BLOCKCACHE => 'true'}, {NAME => 'f3', COMPRESSION => 'NONE'                                       
 , VERSIONS => '3', TTL => '2147483647', BLOCKSIZE => '65536', IN_MEMOR                                       
 Y => 'false', BLOCKCACHE => 'true'}]}                                                                        
1 row(s) in 0.0750 seconds
hbase(main):005:0>

This sums it up. I hope you give HBase on the Cloudera Training VM a whirl as it also has Eclipse installed and therefore provides a quick start into Hadoop and HBase.

Just keep in mind that this is for prototyping only! With such a setup you will only be able to insert a handful of rows. If you overdo it you will bring it to its knees very quickly. But you can safely use it to play around with the shell to create tables or use the API to get used to it and test changes in your code etc.

Finally a screenshot of the running HBase UI:

HBase on Cloudera Training Virtual Machine (0.3.1)

2009-10-20T10:22:00.000-07:00

You might want to run HBase on Cloudera's Virtual Machine to get a quick start to a prototyping setup. In theory you download the VM, start it and you are ready to go. There are a few issues though, the worst being that the current Hadoop Training VM does not include HBase at all. Also, Cloudera is using a specific version of Hadoop that it deems stable and maintains it own release cycle. So Cloudera's version of Hadoop is 0.18.3. HBase though needs Hadoop 0.20 - but we are in luck as Andrew Purtell of TrendMicro maintains a special branch of HBase 0.20 that works with Cloudera's release.

Here are the steps to get HBase running on Cloudera's VM:

Download VM

Get it from Cloudera's website.
Start VM

As the above page states: "To launch the VMWare image, you will either need VMware Player for windows and linux, or VMware Fusion for Mac."

Note: I have Parallels for Mac and wanted to use that. I used Parallels Transporter to convert the "cloudera-training-0.3.1.vmx" to a new "cloudera-training-0.2-cl3-000002.hdd", create a new VM in Parallels selecting Ubuntu Linux as the OS and the newly created .hdd as the disk image. Boot up the VM and you are up and running. I gave it a bit more memory for the graphics to be able to switch the VM to 1440x900 which is native to my MacBook Pro I am using.

Finally follow the steps explained on the page above, i.e. open a Terminal and issue:
```
$ cd ~/git
$ ./update-exercises --workspace
```
Pull HBase branch

Open a new Terminal (or issue a $ cd .. in the open one), then:
```
$ sudo -u hadoop git clone http://git.apache.org/hbase.git /home/hadoop/hbase
$ sudo -u hadoop sh -c "cd /home/hadoop/hbase ; git checkout origin/0.20_on_hadoop-0.18.3"
...
HEAD is now at c050f68... pull up to release
```
First we clone the repository, then switch to the actual branch. You will notice that I am using sudo -u hadoop because Hadoop itself is started under that account and so I wanted it to match. Also, the default "training" account does not have SSH set up as explained in Hadoop's quick-start guide. When sudo is asking for a password use the default set to "training".

Build Branch

Continue in Terminal:

$ sudo -u hadoop sh -c "cd /home/hadoop/hbase/ ; export PATH=$PATH:/usr/share/apache-ant-1.7.1/bin ; ant package"
...
BUILD SUCCESSFUL

Configure HBase

There are a few edits to be made to get HBase running.
```
$ sudo -u hadoop vim /home/hadoop/hbase/build/conf/hbase-site.xml

<configuration>

  <property>
    <name>hbase.rootdir</name>
    <value>hdfs://localhost:8020/hbase</value>
  </property>

</configuration>

$ sudo -u hadoop vim /home/hadoop/hbase/build/conf/hbase-env.sh 

# The java implementation to use.  Java 1.6 required.
# export JAVA_HOME=/usr/java/jdk1.6.0/
export JAVA_HOME=/usr/lib/jvm/java-6-sun
...
```
Note: There is a small glitch in the revision 826669 of that Cloudera specific HBase branch. The master UI (on port 60010 on localhost) will not start because a path is different and Jetty packages are missing because of it. You can fix it by editing the start up script and changing the path scanned:
```
$ sudo -u hadoop vim /home/hadoop/hbase/build/bin/hbase
```
Replace
for f in $HBASE_HOME/lib/jsp-2.1/*.jar; do
with
for f in $HBASE_HOME/lib/jetty-ext/*.jar; do

This is only until the developers have fixed this in the branch (compare the revision I used r813052 with what you get). Or if you do not want the UI you can ignore this and the error in the logs too. HBase will still run, just not its web based interface.

Rev up the Engine!

The final thing is to start HBase:

$ sudo -u hadoop /home/hadoop/hbase/build/bin/start-hbase.sh
$ sudo -u hadoop /home/hadoop/hbase/build/bin/hbase shell

HBase Shell; enter 'help<RETURN>' for list of supported commands.
Version: 0.20.0-0.18.3, r813052, Mon Oct 19 06:51:57 PDT 2009
hbase(main):001:0> list
0 row(s) in 0.2320 seconds
hbase(main):002:0>

Done!

HBase Architecture 101 - Storage

2009-10-12T14:35:00.000-07:00

One of the more hidden aspects of HBase is how data is actually stored. While the majority of users may never have to bother about it you may have to get up to speed when you want to learn what the various advanced configuration options you have at your disposal mean. "How can I tune HBase to my needs?", and other similar questions are certainly interesting once you get over the (at times steep) learning curve of setting up a basic system. Another reason wanting to know more is if for whatever reason disaster strikes and you have to recover a HBase installation.

In my own efforts getting to know the respective classes that handle the various files I started to sketch a picture in my head illustrating the storage architecture of HBase. But while the ingenious and blessed committers of HBase easily navigate back and forth through that maze I find it much more difficult to keep a coherent image. So I decided to put that sketch to paper. Here it is.

Please note that this is not a UML or call graph but a merged picture of classes and the files they handle and by no means complete though focuses on the topic of this post. I will discuss the details below and also look at the configuration options and how they affect the low-level storage files.

The Big Picture

So what does my sketch of the HBase innards really say? You can see that HBase handles basically two kinds of file types. One is used for the write-ahead log and the other for the actual data storage. The files are primarily handled by the HRegionServer's. But in certain scenarios even the HMaster will have to perform low-level file operations. You may also notice that the actual files are in fact divided up into smaller blocks when stored within the Hadoop Distributed Filesystem (HDFS). This is also one of the areas where you can configure the system to handle larger or smaller data better. More on that later.

The general flow is that a new client contacts the Zookeeper quorum (a separate cluster of Zookeeper nodes) first to find a particular row key. It does so by retrieving the server name (i.e. host name) that hosts the -ROOT- region from Zookeeper. With that information it can query that server to get the server that hosts the .META. table. Both of these two details are cached and only looked up once. Lastly it can query the .META. server and retrieve the server that has the row the client is looking for.

Once it has been told where the row resides, i.e. in what region, it caches this information as well and contacts the HRegionServer hosting that region directly. So over time the client has a pretty complete picture of where to get rows from without needing to query the .META. server again.

Note: The HMaster is responsible to assign the regions to each HRegionServer when you start HBase. This also includes the "special" -ROOT- and .META. tables.

Next the HRegionServer opens the region it creates a corresponding HRegion object. When the HRegion is "opened" it sets up a Store instance for each HColumnFamily for every table as defined by the user beforehand. Each of the Store instances can in turn have one or more StoreFile instances, which are lightweight wrappers around the actual storage file called HFile. A HRegion also has a MemStore and a HLog instance. We will now have a look at how they work together but also where there are exceptions to the rule.

Stay Put

So how is data written to the actual storage? The client issues a HTable.put(Put) request to the HRegionServer which hands the details to the matching HRegion instance. The first step is now to decide if the data should be first written to the "Write-Ahead-Log" (WAL) represented by the HLog class. The decision is based on the flag set by the client using Put.writeToWAL(boolean) method. The WAL is a standard Hadoop SequenceFile (although it is currently discussed if that should not be changed to a more HBase suitable file format) and it stores HLogKey's. These keys contain a sequential number as well as the actual data and are used to replay not yet persisted data after a server crash.

Once the data is written (or not) to the WAL it is placed in the MemStore. At the same time it is checked if the MemStore is full and in that case a flush to disk is requested. When the request is served by a separate thread in the HRegionServer it writes the data to an HFile located in the HDFS. It also saves the last written sequence number so the system knows what was persisted so far. Let"s have a look at the files now.

Files

HBase has a configurable root directory in the HDFS but the default is /hbase. You can simply use the DFS tool of the Hadoop command line tool to look at the various files HBase stores.

$ hadoop dfs -lsr /hbase/docs
...
drwxr-xr-x   - hadoop supergroup          0 2009-09-28 14:22 /hbase/.logs
drwxr-xr-x   - hadoop supergroup          0 2009-10-15 14:33 /hbase/.logs/srv1.foo.bar,60020,1254172960891
-rw-r--r--   3 hadoop supergroup      14980 2009-10-14 01:32 /hbase/.logs/srv1.foo.bar,60020,1254172960891/hlog.dat.1255509179458
-rw-r--r--   3 hadoop supergroup       1773 2009-10-14 02:33 /hbase/.logs/srv1.foo.bar,60020,1254172960891/hlog.dat.1255512781014
-rw-r--r--   3 hadoop supergroup      37902 2009-10-14 03:33 /hbase/.logs/srv1.foo.bar,60020,1254172960891/hlog.dat.1255516382506
...
-rw-r--r--   3 hadoop supergroup  137648437 2009-09-28 14:20 /hbase/docs/1905740638/oldlogfile.log
...
drwxr-xr-x   - hadoop supergroup          0 2009-09-27 18:03 /hbase/docs/999041123
-rw-r--r--   3 hadoop supergroup       2323 2009-09-01 23:16 /hbase/docs/999041123/.regioninfo
drwxr-xr-x   - hadoop supergroup          0 2009-10-13 01:36 /hbase/docs/999041123/cache
-rw-r--r--   3 hadoop supergroup   91540404 2009-10-13 01:36 /hbase/docs/999041123/cache/5151973105100598304
drwxr-xr-x   - hadoop supergroup          0 2009-09-27 18:03 /hbase/docs/999041123/contents
-rw-r--r--   3 hadoop supergroup  333470401 2009-09-27 18:02 /hbase/docs/999041123/contents/4397485149704042145
drwxr-xr-x   - hadoop supergroup          0 2009-09-04 01:16 /hbase/docs/999041123/language
-rw-r--r--   3 hadoop supergroup      39499 2009-09-04 01:16 /hbase/docs/999041123/language/8466543386566168248
drwxr-xr-x   - hadoop supergroup          0 2009-09-04 01:16 /hbase/docs/999041123/mimetype
-rw-r--r--   3 hadoop supergroup     134729 2009-09-04 01:16 /hbase/docs/999041123/mimetype/786163868456226374
drwxr-xr-x   - hadoop supergroup          0 2009-10-08 22:45 /hbase/docs/999882558
-rw-r--r--   3 hadoop supergroup       2867 2009-10-08 22:45 /hbase/docs/999882558/.regioninfo
drwxr-xr-x   - hadoop supergroup          0 2009-10-09 23:01 /hbase/docs/999882558/cache
-rw-r--r--   3 hadoop supergroup   45473255 2009-10-09 23:01 /hbase/docs/999882558/cache/974303626218211126
drwxr-xr-x   - hadoop supergroup          0 2009-10-12 00:37 /hbase/docs/999882558/contents
-rw-r--r--   3 hadoop supergroup  467410053 2009-10-12 00:36 /hbase/docs/999882558/contents/2507607731379043001
drwxr-xr-x   - hadoop supergroup          0 2009-10-09 23:02 /hbase/docs/999882558/language
-rw-r--r--   3 hadoop supergroup        541 2009-10-09 23:02 /hbase/docs/999882558/language/5662037059920609304
drwxr-xr-x   - hadoop supergroup          0 2009-10-09 23:02 /hbase/docs/999882558/mimetype
-rw-r--r--   3 hadoop supergroup      84447 2009-10-09 23:02 /hbase/docs/999882558/mimetype/2642281535820134018
drwxr-xr-x   - hadoop supergroup          0 2009-10-14 10:58 /hbase/docs/compaction.dir

The first set of files are the log files handled by the HLog instances and which are created in a directory called .logs underneath the HBase root directory. Then there is another subdirectory for each HRegionServer and then a log for each HRegion.

Next there is a file called oldlogfile.log which you may not even see on your cluster. They are created by one of the exceptions I mentioned earlier as far as file access is concerned. They are a result of so called "log splits". When the HMaster starts and finds that there is a log file that is not handled by a HRegionServer anymore it splits the log copying the HLogKey's to the new regions they should be in. It places them directly in the region's directory in a file named oldlogfile.log. Now when the respective HRegion is instantiated it reads these files and inserts the contained data into its local MemStore and starts a flush to persist the data right away and delete the file.

Note: Sometimes you may see left-over oldlogfile.log.old (yes, there is another .old at the end) which are caused by the HMaster trying repeatedly to split the log and found there was already another split log in place. At that point you have to consult with the HRegionServer or HMaster logs to see what is going on and if you can remove those files. I found at times that they were empty and therefore could safely be removed.

The next set of files are the actual regions. Each region name is encoded using a Jenkins Hash function and a directory created for it. The reason to hash the region name is because it may contain characters that cannot be used in a path name in DFS. The Jenkins Hash always returns legal characters, as simple as that. So you get the following path structure:

/hbase/<tablename>/<encoded-regionname>/<column-family>/<filename>

In the root of the region directory there is also a .regioninfo holding meta data about the region. This will be used in the future by an HBase fsck utility (see HBASE-7) to be able to rebuild a broken .META. table. For a first usage of the region info can be seen in HBASE-1867.

In each column-family directory you can see the actual data files, which I explain in the following section in detail.

Something that I have not shown above are split regions with their initial daughter reference files. When a data file within a region grows larger than the configured hbase.hregion.max.filesize then the region is split in two. This is done initially very quickly because the system simply creates two reference files in the new regions now supposed to host each half. The name of the reference file is an ID with the hashed name of the referenced region as a postfix, e.g. 1278437856009925445.3323223323. The reference files only hold little information: the key the original region was split at and wether it is the top or bottom reference. Of note is that these references are then used by the HalfHFileReader class (which I also omitted from the big picture above as it is only used temporarily) to read the original region data files. Only upon a compaction the original files are rewritten into separate files in the new region directory. This also removes the small reference files as well as the original data file in the original region.

And this also concludes the file dump here, the last thing you see is a compaction.dir directory in each table directory. They are used when splitting or compacting regions as noted above. They are usually empty and are used as a scratch area to stage the new data files before swapping them into place.

HFile

So we are now at a very low level of HBase's architecture. HFile's (kudos to Ryan Rawson) are the actual storage files, specifically created to serve one purpose: store HBase's data fast and efficiently. They are apparently based on Hadoop's TFile (see HADOOP-3315) and mimic the SSTable format used in Googles BigTable architecture. The previous use of Hadoop's MapFile's in HBase proved to be not good enough performance wise. So how do the files look like?

The files have a variable length, the only fixed blocks are the FileInfo and Trailer block. As the picture shows it is the Trailer that has the pointers to the other blocks and it is written at the end of persisting the data to the file, finalizing the now immutable data store. The Index blocks record the offsets of the Data and Meta blocks. Both the Data and the Meta blocks are actually optional. But you most likely you would always find data in a data store file.

How is the block size configured? It is driven solely by the HColumnDescriptor which in turn is specified at table creation time by the user or defaults to reasonable standard values. Here is an example as shown in the master web based interface:

{NAME => 'docs', FAMILIES => [{NAME => 'cache', COMPRESSION => 'NONE', VERSIONS => '3', TTL => '2147483647', BLOCKSIZE => '65536', IN_MEMORY => 'false', BLOCKCACHE => 'false'}, {NAME => 'contents', COMPRESSION => 'NONE', VERSIONS => '3', TTL => '2147483647', BLOCKSIZE => '65536', IN_MEMORY => 'false', BLOCKCACHE => 'false'}, ...

The default is "64KB" (or 65535 bytes). Here is what the HFile JavaDoc explains:

"Minimum block size. We recommend a setting of minimum block size between 8KB to 1MB for general usage. Larger block size is preferred if files are primarily for sequential access. However, it would lead to inefficient random access (because there are more data to decompress). Smaller blocks are good for random access, but require more memory to hold the block index, and may be slower to create (because we must flush the compressor stream at the conclusion of each data block, which leads to an FS I/O flush). Further, due to the internal caching in Compression codec, the smallest possible block size would be around 20KB-30KB."

So each block with its prefixed "magic" header contains either plain or compressed data. How that looks like we will have a look at in the next section.

One thing you may notice is that the default block size for files in DFS is 64MB, which is 1024 times what the HFile default block size is. So the HBase storage files blocks do not match the Hadoop blocks. Therefore you have to think about both parameters separately and find the sweet spot in terms of performance for your particular setup.

One option in the HBase configuration you may see is hfile.min.blocksize.size. It seems to be only used during migration from earlier versions of HBase (since it had no block file format) and when directly creating HFile during bulk imports for example.

So far so good, but how can you see if a HFile is OK or what data it contains? There is an App for that!

The HFile.main() method provides the tools to dump a data file:

$ hbase org.apache.hadoop.hbase.io.hfile.HFile
usage: HFile  [-f ] [-v] [-r ] [-a] [-p] [-m] [-k]
 -a,--checkfamily    Enable family check
 -f,--file      File to scan. Pass full-path; e.g.
                     hdfs://a:9000/hbase/.META./12/34
 -k,--checkrow       Enable row order check; looks for out-of-order keys
 -m,--printmeta      Print meta data of file
 -p,--printkv        Print key/value pairs
 -r,--region    Region to scan. Pass region name; e.g. '.META.,,1'
 -v,--verbose        Verbose output; emits file and meta data delimiters

Here is an example of what the output will look like (shortened here):

$ hbase org.apache.hadoop.hbase.io.hfile.HFile -v -p -m -f \
  hdfs://srv1.foo.bar:9000/hbase/docs/999882558/mimetype/2642281535820134018

Scanning -> hdfs://srv1.foo.bar:9000/hbase/docs/999882558/mimetype/2642281535820134018
...
K: \x00\x04docA\x08mimetype\x00\x00\x01\x23y\x60\xE7\xB5\x04 V: text\x2Fxml
K: \x00\x04docB\x08mimetype\x00\x00\x01\x23x\x8C\x1C\x5E\x04 V: text\x2Fxml
K: \x00\x04docC\x08mimetype\x00\x00\x01\x23xz\xC08\x04 V: text\x2Fxml
K: \x00\x04docD\x08mimetype\x00\x00\x01\x23y\x1EK\x15\x04 V: text\x2Fxml
K: \x00\x04docE\x08mimetype\x00\x00\x01\x23x\xF3\x23n\x04 V: text\x2Fxml
Scanned kv count -> 1554

Block index size as per heapsize: 296
reader=hdfs://srv1.foo.bar:9000/hbase/docs/999882558/mimetype/2642281535820134018, \
  compression=none, inMemory=false, \
  firstKey=US6683275_20040127/mimetype:/1251853756871/Put, \
  lastKey=US6684814_20040203/mimetype:/1251864683374/Put, \
  avgKeyLen=37, avgValueLen=8, \
  entries=1554, length=84447
fileinfoOffset=84055, dataIndexOffset=84277, dataIndexCount=2, metaIndexOffset=0, \
  metaIndexCount=0, totalBytes=84055, entryCount=1554, version=1
Fileinfo:
MAJOR_COMPACTION_KEY = \xFF
MAX_SEQ_ID_KEY = 32041891
hfile.AVG_KEY_LEN = \x00\x00\x00\x25
hfile.AVG_VALUE_LEN = \x00\x00\x00\x08
hfile.COMPARATOR = org.apache.hadoop.hbase.KeyValue\x24KeyComparator
hfile.LASTKEY = \x00\x12US6684814_20040203\x08mimetype\x00\x00\x01\x23x\xF3\x23n\x04

The first part is the actual data stored as KeyValue pairs, explained in detail in the next section. The second part dumps the internal HFile.Reader properties as well as the Trailer block details and finally the FileInfo block values. This is a great way to check if a data file is still healthy.

KeyValue's

In essence each KeyValue in the HFile is simply a low-level byte array that allows for "zero-copy" access to the data, even with lazy or custom parsing if necessary. How are the instances arranged?

The structure starts with two fixed length numbers indicating the size of the key and the value part. With that info you can offset into the array to for example get direct access to the value, ignoring the key - if you know what you are doing. Otherwise you can get the required information from the key part. Once parsed into a KeyValue object you have getters to access the details.

Note: One thing to watch out for is the difference between KeyValue.getKey() and KeyValue.getRow(). I think for me the confusion arose from referring to "row keys" as the primary key to get a row out of HBase. That would be the latter of the two methods, i.e. KeyValue.getRow(). The former simply returns the complete byte array part representing the raw "key" as colored and labeled in the diagram.

This concludes my analysis of the HBase storage architecture. I hope it provides a starting point for your own efforts to dig into the grimy details. Have fun!

Update: Slightly updated with more links to JIRA issues. Also added Zookeeper to be more precise about the current mechanisms to look up a region.

Update 2: Added details about region references.

Update 3: Added more details about region lookup as requested.

Hive vs. Pig

2009-10-12T13:20:00.000-07:00

While I was looking at Hive and Pig for processing large amounts of data without the need to write MapReduce code I found that there is no easy way to compare them against each other without reading into both in greater detail.

In this post I am trying to give you a 10,000ft view of both and compare some of the more prominent and interesting features. The following table - which is discussed below - compares what I deemed to be such features:

Feature	Hive	Pig
Language	SQL-like	PigLatin
Schemas/Types	Yes (explicit)	Yes (implicit)
Partitions	Yes	No
Server	Optional (Thrift)	No
User Defined Functions (UDF)	Yes (Java)	Yes (Java)
Custom Serializer/Deserializer	Yes	Yes
DFS Direct Access	Yes (implicit)	Yes (explicit)
Join/Order/Sort	Yes	Yes
Shell	Yes	Yes
Streaming	Yes	Yes
Web Interface	Yes	No
JDBC/ODBC	Yes (limited)	No

Let us look now into each of these with a bit more detail.

General Purpose

The question is "What does Hive or Pig solve?". Both - and I think this lucky for us in regards to comparing them - have a very similar goal. They try to ease the complexity of writing MapReduce jobs in a programming language like Java by giving the user a set of tools that they may be more familiar with (more on this below). The raw data is stored in Hadoop's HDFS and can be any format although natively it usually is a TAB separated text file, while internally they also may make use of Hadoop's SequenceFile file format. The idea is to be able to parse the raw data file, for example a web server log file, and use the contained information to slice and dice them into what is needed for business needs. Therefore they provide means to aggregate fields based on specific keys. In the end they both emit the result again in either text or a custom file format. Efforts are also underway to have both use other systems as a source for data, for example HBase.

The features I am comparing are chosen pretty much at random because they stood out when I read into each of these two frameworks. So keep in mind that this is a subjective list.

Language

Hive lends itself to SQL. But since we can only read already existing files in HDFS it is lacking UPDATE or DELETE support for example. It focuses primarily on the query part of SQL. But even there it has its own spin on things to reflect better the underlaying MapReduce process. Overall is seems that someone familiar with SQL can very quickly learn Hive's version of it and get results fast.

Pig on the other hand looks more like a very simplistic scripting language. As with those (and this is a nearly religious topic) some are more intuitive and some are less. As with PigLatin I was able to see what the samples do, but lacking the full knowledge of its syntax I was somewhat finding myself thinking if I really would be able to get what I needed without too many trial-and-error loops. Sure, the Hive SQL needs probably as many iterations to fully grasp - but there is at least a greater understanding of what to expect.

Schemas/Types

Hive uses once more a specific variation of SQL's Data Definition Language (DDL). It defines the "tables" beforehand and stores the schema in a either shared or local database. Any JDBC offering will do, but it also comes with a built in Derby instance to get you started quickly. If the database is local then only you can run specific Hive commands. If you share the database then others can also run these - or would have to set up their own local database copy. Types are also defined upfront and supported types are INT, BIGINT, BOOLEAN, STRING and so on. There are also array types that lets you handle specific fields in the raw data files as a group.

Pig has no such metadata database. Datatypes and schemas are defined within each script. Types furthermore are usually automatically determined by their use. So if you use a field as an Integer it is handled that way by Pig. You do have the option though to override it and have explicit type definitions, again within the script you need them. Pig has a similar set of types compared to Hive. For example it also has an array type called "bag".

Partitions

Hive has a notion of partitions. They are basically subdirectories in HDFS. It allows for example processing a subset of the data by alphabet or date. It is up to the user to create these "partitions" as they are not enforced nor required.

Pig does not seem to have such a feature. It may be that filters can achieve the same but it is not immediately obvious to me.

Server

Hive can start an optional server, which is allegedly Thrift based. With the server I presume you can send queries from anywhere to the Hive server which in turn executes them.

Pig does not seem to have such a facility yet.

User Defined Functions

Hive and Pig allow for user functionality by supplying Java code to the query process. These functions can add any additional feature that is required to crunch the numbers as required.

Custom Serializer/Deserializer

Again, both Hive and Pig allow for custom Java classes that can read or write any file format required. I also assume that is how it connects to HBase eventually (just a guess). You can write a parser for Apache log files or, for example, the binary Tokyo Tyrant Ulog format. The same goes for the output, write a database output class and you can write the results back into a database.

DFS Direct Access

Hive is smart about how to access the raw data. A "select * from table limit 10" for example does a direct read from the file. If the query is too complicated it will fall back to use a full MapReduce run to determine the outcome, just as expected.

With Pig I am not sure if it does the same to speed up simple PigLatin scripts. At least it does not seem to be mentioned anywhere as an important feature.

Join/Order/Sort

Hive and Pig have support for joining, ordering or sorting data dynamically. They perform the same purpose in both pretty allowing you to aggregate and sort the result as is needed. Pig also has a COGROUP feature that allows you to do OUTER JOIN's and so on. I think this is where you spent most of your time with either package - especially when you start out. But from a cursory look it seems both can do pretty much the same.

Shell

Both Hive and Pig have a shell that allows you to query specific things or run the actual queries. Pig also passes on DFS commands such as "cat" to allow you to quickly check what an outcome of a specific PigLatin script was.

Streaming

Once more, both frameworks seem to provide streaming interfaces so that you can process data with external tools or languages, such as Ruby or Python. How the streaming performs I do not know and if they affect them differently. This is for you to tell me :)

Web Interface

Only Hive has a web interface or UI that can be used to visualize the various schemas and issue queries. This is different to the above mentioned Server as it is an interactive web UI for a human operator. The Hive Server is for use from another programming or scripting language for example.

JDBC/ODBC

Another Hive only feature is the availability of a - again limited functionality - JDBC/ODBC driver. It is another way for programmers to use Hive without having to bother with its shell or web interface, or even the Hive Server. Since only a subset of features is available it will require small adjustments on the programmers side of things but otherwise seems like a nice-to-have feature.

Conclusion

Well, it seems to me that both can help you achieve the same goals, while Hive comes more natural to database developers and Pig to "script kiddies" (just kidding). Hive has more features as far as access choices are concerned. They also have reportedly roughly the same amount of committers in each project and are going strong development wise.

This is it from me. Do you have a different opinion or comment on the above then please feel free to reply below. Over and out!

HBase Schema Manager

2009-05-26T05:12:00.000-07:00

As already mentioned in one of my previous posts, HBase at times makes it difficult to maintain or even create a new table structure. Imagine you have a running cluster and quite an elaborate table setup. Now you want to to create a backup cluster for load balancing and general tasks like reporting etc. How do you get all the values from one system into the other?

While you can use various examples that help you backing up the data and eventually restore it, how do you "clone" the table schemas?

Or imagine you have an existing system like the one we talked about above and you simply want to change a few things around. With an RDBMS you can save the required steps in a DDL statement and execute it on the server - or the backup server etc. But with HBase there is now DDL or even the possibility of executing pre-built scripts against a running cluster.

What I described in my previous post was a why to store the table schemas into an XML configuration file and run that against a cluster. The code handles adding new tables and more importantly the addition, removal and modification of column families for any named table.

I have put this all into a separate Java application that may be useful to you. You can get it from my GitHub repository. It is really simple to use, you create an XML based configuration file, for example:

<?xml version="1.0" encoding="UTF-8"?>
<configurations>
  <configuration>
    <name>foo</name>
    <description>Configuration for the FooBar HBase cluster.</description>
    <hbase_master>foo.bar.com:60000</hbase_master>
    <schema>
      <table>
        <name>test</name>
        <description>Test table.</description>
        <column_family>
          <name>sample</name>
          <description>Sample column.</description>
          <!-- Default: 3 -->
          <max_versions>1</max_versions>
          <!-- Default: DEFAULT_COMPRESSION_TYPE -->
          <compression_type/>
          <!-- Default: false -->
          <in_memory/>
          <!-- Default: false -->
          <block_cache_enabled/>
          <!-- Default: -1 (forever) -->
          <time_to_live/>
          <!-- Default: 2147483647 -->
          <max_value_length/>
          <!-- Default: DEFAULT_BLOOM_FILTER_DESCRIPTOR -->
          <bloom_filter/>
        </column_family>
      </table>
    </schema>
  </configuration>
</configurations>

Then all you have to do is run the application like so:

java -jar hbase-manager-1.0.jar schema.xml

The "schema.xml" is the above XML configuration saved on your local machine. The output shows the steps performed:

$ java -jar hbase-manager-1.0.jar schema.xml
 creating table test...
 table created
done.

You can also specify more options on the command line:

usage: HbaseManager [<options>] <schema-xml-filename> [<config-name>]
 -l,--list       lists all tables but performs no further action.
 -n,--nocreate   do not create non-existent tables.
 -v,--verbose    print verbose output.

If you use the "verbose" option you get more details:

$ java -jar hbase-manager-1.0.jar -v schema.xml
schema filename: schema.xml
configuration used: null
using config number: default
table schemas read from config: 
  [name -> test
  description -> Test table.
  columns -> {sample=name -> sample
  description -> Sample column.
  maxVersions -> 1
  compressionType -> NONE
  inMemory -> false
  blockCacheEnabled -> false
  maxValueLength -> 2147483647
  timeToLive -> -1
  bloomFilter -> false}]
 hbase.master -> foo.bar.com:60000
 authoritative -> true
 name -> test
 tableExists -> true
 changing table test...
 no changes detected!
done.

Finally you can use the "list" option to check initial connectivity and the successful changes:

$ java -jar hbase-manager-1.0.jar -l schema.xml
tables found: 1
  test
done.

A few notes: First and most importantly, if you change a large table, i.e. one with thousands of regions, this process can take quite a long time. This is caused by the enableTable() call having to scan the complete .META. table to assign the regions to their respective region servers. There is possibly room for improvement in my little application to handle this better - suggestions welcome!

Also, I do not have Bloom Filter settings implemented, as this is still changing from 0.19 to 0.20. Once it has been finalized I will add support for it.

If you do not specify a configuration name then the first one is used. Having more than one configuration allows you to have multiple clusters defined in one schema file and by specifying the name you can execute only a specific one when you need to.

HBase MapReduce 101 - Part I

2009-05-11T15:08:00.000-07:00

In this and the following posts I would like to take the opportunity to go into detail about the MapReduce process as provided by Hadoop but more importantly how it applies to HBase.

MapReduce

MapReduce as a process was designed to solve the problem of processing in excess of terabytes of data in a scalable way. There should be a way to design such a system that increases in performance linearly with the number of physical machines added. That is what MapReduce strives to do. It follows a divide-and-conquer approach by splitting the data located on a distributed file system so that the servers (or rather cpu's, or more modern "cores") available can access these pieces and process them as fast as they can. The problem with this approach is that you will have to consolidate the data at the end. Again, MapReduce has this built right into it.

The above simplified image of the MapReduce process shows you how the data is processed. The first thing that happens is the split which is responsible to divide the input data in reasonable size chunks that are then processed by one server at a time. This splitting has to be done somewhat smart to make best use of available servers and the infrastructure in general. In this example the data may be a very large log file that is divided into equal size pieces on line boundaries. This is OK for example for say Apache log files. Input data may also be binary though where you may have to write your own getSplits() method - but more on that below.

Classes

The above image also shows you the classes that are involved in the Hadoop implementation of MapReduce. Let's look at them and also at the specific implementations that HBase provides on top of those.

InputFormat

The first class to deal with is the InputFormat class. It is responsible for two things. First is does the actual splitting of the input data as well as returning a RecordReader instance that defines the classes of the key and value objects as well as providing a next() method that is used to iterate over each input record.

As far as HBase is concerned there is a special implementation called TableInputFormatBase as well as its subclass TableInputFormat. The former implements the majority of the functionality but remains abstract. The subclass is a light-weight concrete version of the TableInputFormat and is used by many supplied sample and real MapReduce classes.

But most importantly these classed implement the full turn-key solution to scan a HBase table. You can provide the name of the table to scan and the columns you want to process during the Map phase. It splits the table into proper pieces for you and hands them over to the subsequent classes. There are quite a few tweaks which we will address below and in the following installments of this series.

For now let's look at the other classes involved.

Mapper

The Mapper class(es) are for the next stage of the MapReduce process and one of its namesakes. In this step each record read using the RecordReader is processed using the map() method. What is also visible somewhat from the first figure above is that the Mapper reads a specific type of key/value pair but emits possibly another. This is handy to convert the raw data into something more useful for further processing.

Again, looking at HBase's extensions to this, you will find a TableMap class that is specific to iterating over a HBase table. Once specific implementation is the IdentityTableMap which is also a good example on how to add your own functionality to the supplied classes. The TableMap class itself does not implement anything but only adds the signatures of what the actual key/value pair classes are. The IdentityTableMap is simply passing on the records to the next stage of the processing.

Reducer

The Reduce stage and class layout is very similar to the Mapper one explained above. This time we get the output of a Mapper class and process it after the data was shuffled and sorted. In the implicit shuffle between the Mapper and Reducer stages the intermediate data is copied from different Map to the Reduce servers and the sort combines the shuffled (copied) data so that the Reducer sees the intermediate data as a nicely sorted set where now each unique key (and that is something I will get back to later) is associated with all of the possible values it was found with.

OutputFormat

The final stage is the OutputFormat class and its job to persist the data in various locations. There are specific implementations that allow output to files or to HBase tables in case of the TableOutputFormat. It uses a RecordWriter to write the data into the specific HBase output table.

It is important to note the cardinality as well. While there are many Mappers handing records to many Reducers, there is only one OutputFormat that takes each output record from its Reducer subsequently. It is the final class handling the key/value pairs and writes them to their final destination, this being a file or a table.

The name of the output table is specified when the job is created. Otherwise it does not add much more complexity. One rather significant thing it does is set the table's auto flush to "false" and handles the buffer flushing implicitly. This helps a lot speeding up the import of large data sets.

To Map or Reduce or Not Map or Reduce

This is now a crucial point we are at deciding on how to process tables stored in HBase. From the above it seems that we simply use a TableInputFormat to feed through a s TableMap and TableReduce to eventually persist the data with a TableOutputFormat. But this may not be what you want when you deal with HBase. The question is if there is a better way to handle the process given certain specific architectural features HBase provides. Depending on your data source and target there are a few different scenarios we could think of.

If you want to import a large set of data into a HBase table you can read the data using a Mapper and after aggregating it on a per key basis using a Reducer and finally writing it into a HBase table. This involves the whole MapReduce stack including the shuffle and sort using intermediate files. But what if you know that the data for example has a unique key? Why go through the extra step of copying and sorting when there is always just exactly one key/value pair? At this point you can ask yourself, wouldn't it be better if I could skip that whole reduce stage? The answer is yes you can! And you should as you can harvest the pure computational power of all CPU's to crunch the data and writing it at top IO speed to its final target.

As you can see from the matrix, there are quite a few scenarios where you can decide if you want Map only or both, Map and Reduce. But when it comes to handling HBase tables as sources and targets there are a few exceptions to this rule and I highlighted them accordingly.

Same or different Tables

This is the an important distinction. The bottom line is, when you read a table in the Map stage you should consider not writing back to that very same table in the same process. It could on one hand hinder the proper distribution of regions across the servers (open scanners block regions splits) and on the other hand you may or may not see the new data as you scan. But when you read from one table and write to another then you can do that in a single stage. So for two different tables you can write your table updates directly in the TableMap.map() while with the same table you must write the same code in the TableReduce.reduce() - or in its TableOutputFormat (or even better simply use that class as is and you are done). The reason is that the Map stage completely reads a table and then passes the data on in intermediate files to the Reduce stage. In turn this means that the Reducer reads from the distributed file system (DFS) and writes into the now idle HBase table. And all is well.

All of the above are simply recommendations based on what is currently available with HBase. There is of course no reason not to scan and modify a table in the same process. But to avoid certain non-deterministic issue I personally would not recommend this - especially if you are new to HBase. Maybe this also could be taken into consideration when designing your HBase tables. Maybe you separate distinct data into two tables or separate column families so you can scan one table while changing the other.

Key Distribution

Another specific requirement for an effective import is to have a random keys as they are read. While this will be difficult if you scan a table in the Map phase, as keys are sorted, you may be able to make use of this when reading from a raw data file. Instead of leaving the key the offset of the file, as created by the TextOutputFormat for example, you could simply replace the rather useless offset with a random key. This will guarantee that the data is spread across all servers more evenly. Especially the HRegionServers will be very thankful as they each host as set of regions and random keys makes for a random load on these regions.

Of course this depends on how the data is written to the raw files or how the real row keys are computed, but still a very valuable thing to keep in mind.

In the next post I will show you how to import data from a raw data file into a HBase table and how you eventually process the data in the HBase table. We will address questions like how many mappers and/or reducers are needed and how can I improve import and processing performance.

Until then, have a good day!

European HBase Ambassador

2009-05-03T11:09:00.000-07:00

I am on a mission! The mission is to spread the word on HBase. There are only a few choices when it comes to large scale data storage solutions. What I am referring to is not your vertical scale, big-iron relational database system. The time is now for a paradigm shift.

There are various key/value or more structured stores that strive to achieve the same - but getting proper information is often a big challenge. Either there is no proper examples, or use-cases for that matter. Even worse are performance details, the usual rap being "all depends!" - of course it does. Without real examples it is difficult to determine if a suitable system design wise is also holding up to the task at hand. How many servers are needed? How should the hardware stack be organized?

Here is where I feel I can help. I am using HBase for over a year now (started prototyping in late 2007), in production with three clusters spread over more than 50 servers. And I pretty much set them up from plugging in the hardware to the designing and running the cluster and the system on top of that. While this is in itself nothing special, I feel that I gained a lot of experience with HBase. I also feel that it could be very helpful to others that are thinking about HBase and how it may help them.

So I hereby declare myself to be a European HBase Ambassador. Mind you, this is no official title. The purpose is to help furthering the adoption of HBase in research and/or commercial projects. So what does this "position" entail? I offer this:

For the cost of travel (and if necessary accommodation) I will present any aspect of HBase in production to whoever is interested.

Yes, I will come to you if you ask, no matter where in Europe (or beyond). I will present on the internals of HBase, its API and so on to developers or higher concepts to architects. Or to management on a white paper level. You want to know about HBase and how it can help you? Let me know and I will show you.

Why me? Besides what I mentioned above I have over 13 years experience in software engineering and am responsible as the CTO at WorldLingo which for example is the sole provider for all machine translations in Microsoft Office for Windows and MacOS - given the text is longer than a few words because otherwise the internal word dictionaries are taking preference. I write and speak German (native) and English fluently. Last but not least I have regular contact with the core developers of HBase and am a contributor myself - as much as time allows.

So there you have it. This is my offer. I hope you see value in and take me up on it! I surely am looking forward to meeting you.

10 Years in one Project

2009-03-31T15:04:00.001-07:00

For about ten years now I am the CTO at WorldLingo. During those years I have seen quite a few people join and leaving us eventually. Below is a small snapshot of how time has passed. Obviously I am quite proud to be somewhat the rock in the sea.

If you like to know how the video was created, then read on.

I download the source of the code_swarm project following the description, i.e. I used


svn checkout http://codeswarm.googlecode.com/svn/trunk/ codeswarm-read-only
cd codeswarm-read-only
ant all

to get the code and then ran ant all in its root directory:


C:\CODESW~1>ant all
Buildfile: build.xml

init:
     [echo] Running INIT

build:
     [echo] Running BUILD
    [mkdir] Created dir: C:\CODESW~1\build
    [javac] Compiling 18 source files to C:\CODESW~1\build
     [copy] Copying 1 file to C:\CODESW~1\build

jar:
     [echo] Running JAR
    [mkdir] Created dir: C:\CODESW~1\dist
      [jar] Building jar: C:\CODESW~1\dist\code_swarm.jar

all:
     [echo] Building ALL

BUILD SUCCESSFUL
Total time: 6 seconds

Note that this is on my Windows machine. After the build you will have to edit the config file to have your settings and regular expressions match your project. I really took the supplied sample config file, copied it and modified these lines:


# This is a sample configuration file for code_swarm

...

# Input file
InputFile=data/wl-repevents.xml

...

# Project time per frame
#MillisecondsPerFrame=51600000

...

# Optional Method instead of MillisecondsPerFrame
FramesPerDay=2

...

ColorAssign1="wlsystem",".*wlsystem.*", 0,0,255, 0,0,255
ColorAssign2="www",".*www.*", 0,255,0, 0,255,0
ColorAssign3="docs",".*docs.*", 102,0,255, 102,0,255
ColorAssign4="serverconfig",".*serverconf.*", 255,0,0, 255,0,0

# Save each frame to an image?
TakeSnapshots=true

...

DrawNamesHalos=true

...

This is just adjusting the labels and turning on the snap shots to be able to create a video at the end. I found a tutorial that explained how to set this up.

What did not work for me is getting mencoder to work. I downloaded the MPlayer Windows installer from its official site and although it is meant to have mencoder included it does not. Or I am blind.

So, I simply ran

mkdir frames
runrepositoryfetch.bat data\wl.config

to fetch the history of our repository spanning about 10 years - going from Visual SourceSafe, to CVS and currently running on Subversion. One further problem was that the output file of the above script was not named as I had previously specified in the config file, so I had to rename it like so:

cd data
ren realtime_sample1157501935.xml wl-repevents.xml

After that I was able to use run.bat data\wl.config to see the full movie in real time.

With the snap shots created but me not willing to further dig into the absence of mencoder I fired up my trusted MacBookPro and used Quicktime to create the movie from an image sequence.

When Quicktime did its magic I saved the .mov file and used VisualHub to convert it to a proper video format to upload to Vimeo. And that was it really.

CouchDB and CouchApp

2009-03-16T13:11:00.000-07:00

As mentioned in my previous post, I wanted to see CouchDB in action and decided to "push" one of the available sample applications into it. CouchDB has a built in web server to support the REST based API CouchDB supports. The developers were smart enough to see its broader use and allow for applications to be uploaded into a database. An application is a set of static HTML files, images, and JavaScripts that can form a fully functional web application, including the data stored directly in that very same database. With CouchDB's built in replication you get a fully distributed application. Sweet!

Sure you will have to install a load balancer in front of multiple instances of CouchDB, but that is a simple engineering task. I am not sure how session handling will work though. Without a somewhat standard session handling framework it may be difficult to build state aware applications. You can save the session in the database of course and reload upon each request. Is it replicated fast enough though for random access to cluster nodes?

Back to the sample application. I chose the Twitter client provided by Chris Anderson from the CouchDB team (btw, his blog is now hosted directly on CouchDB). I got the tar ball from the above GitHub repository and unpacked it. To get it "pushed" into a CouchDB database you need the actual CouchApp as well. Download its tar ball as well and unpack it. Now we can install CouchApp first. I went with the README file provide and tried the Ruby version, installing it as a gem:


$ sudo gem update --system
$ sudo gem install couchapp

While that installed fine, the syntax that you then find for example for the application to be pushed into the database does not match with the Ruby version of CouchApp. After talking to Chris on the IRC channel he advised uninstalling the Ruby version and rather use the Python based version. I uninstalled the gem and ran:


$ sudo easy_install couchapp

Now the syntax of couchapp did match and I was ready to upload the Twitter sample application - or was I?

Not so fast though, every attempt to upload the application resulted in partial failures. I compared what I had in the uploaded database with what Chris had. His had the attachments needed, mine did not. The database has been created but half of the files were missing. After a quick chat with Chris again we realized that he is using the trunk version of CouchDB, I was using the latest release. Mine was simply outdated and was missing the new application extensions. Here are the next steps I had to run:


$ cd /downloads
$ svn co http://svn.apache.org/repos/asf/couchdb/trunk couchdb
$ cd couchdb/
$ ./bootstrap
$ ./configure
$ make && sudo make install
$ sudo -i -u couchdb couchdb

and to push the application


$ cd ../jchris-couchdb-twitter-client-6bee14ae1b3525d56d77dd9c114002582dc0abe8
$ couchapp push http://localhost:5984/test-twitter

As expected, the application push did succeed now. I was able to see all the files in the newly created database. The screen shot shows the design view of the new "test-twitter" database. Everything you need to serve the application is there, even the favicon.ico to display in the browsers address bar. If you click on the "index.html" the newly uploaded application is started and after logging into Twitter I had everything running as I wanted it.

Here is the a screen shot of the application running. This is really great. I just have to figure out now for myself how to use it either for work or privately. So many choices - but only 24 hours in a day.

Erlang and CouchDB on MacOS

2009-03-15T07:28:00.000-07:00

As I mentioned before I am looking into how I could use Erlang in our own efforts. Sure, it is not the silver bullet for all problems ("Can it make coffee?") and all the hype on the developer ether. But it is certainly built to be the basis of large concurrent systems, for example Facebook's chat system. Or the below mentioned Amazon Dynamo clone called Dynomite.

For starters I purchased "The Pragmatic Programmers" screen case Erlang in Practise with Kevin Smith. I have to say, it is worth every cent! I did watch it on a flight from Munich to Las Vegas and it made the hours literally "fly by". There is something really cool about seeing a program being developed in front of your eyes and re-written many times to explain more advanced concepts as you go along. Highly recommended.

Now I have a background in Prolog and Lisp so getting into the Erlang way of doing things was not too difficult. Of course, the difficult yet again is how to use it for something own. I decided to first build Erlang on my MacBook Pro and then try an Erlang based system to see it working. Building Erlang on MacOS is described in here but overall it is a simple "wget && tar -zxvf && ./configure && make" obstacle course with a few hickups thrown in for good measure. As the post describes, you first need to install XCode from either the supplied MacOS disks or by downloading it from the net - easy peasy. Next is to build libgd. The above post has a link to the details required to build libgd on MacOS. It first required downloading all the required library tar balls and then running the following commands:


$ cd /downloads/
$ tar -zxvf zlib-1.2.3.tar.gz 
$ tar -zxvf gd-2.0.35.tar.gz 
$ tar -zxvf freetype-2.3.8.tar.gz 
$ tar -zxvf jpegsrc.v6b.tar.gz 
$ tar -zxvf libpng-1.2.34.tar.gz

This assumes all tar balls are saved in the "/downloads" directory. Next is zlib:


$ cd zlib-1.2.3 ; ./configure --shared && make && sudo make install
$ ./example

Then libpng:


$ cd ../libpng-1.2.34
$ cp scripts/makefile.darwin Makefile
$ vim Makefile
$ make && sudo make install
$ export srcdir=.; ./test-pngtest.sh

Next is the jpeg library. Here I did not have to symlink the libtool as described in the post. I assume that is because I am on MacOS 10.5. So all I did is this:


$ cd ../jpeg-6b/
$ cp /usr/share/libtool/config.sub .
$ cp /usr/share/libtool/config.guess .
$ ./configure --enable-shared
$ make
$ sudo make install
$ sudo ranlib /usr/local/lib/libjpeg.a

We are getting closer. The freetype library needs these steps, where the last line is for the subsequent libgd build:


$ cd ../freetype-2.3.8
$ ./configure && make && sudo make install
$ sudo ln -s /usr/X11R6/include/fontconfig /usr/local/include

OK, now the libgd library:


$ cd ../gd-2.0.35
$ ln -s `which glibtool` ./libtool
$ ./configure 
$ make && sudo make install 
$ ./gdtest test/gdtest.png

With this all done we can build Erlang from sources like so:


$ cd /downloads
$ tar -zxvf otp_src_R12B-5.tar.gz 
$ cd otp_src_R12B-5
$ ./configure --enable-hipe --enable-smp-support --enable-threads
$ make && sudo make install
$ erl

All cool, the Erlang shell starts up and is ready for action:


$ erl
Erlang (BEAM) emulator version 5.6.5 [source] [smp:2] [async-threads:0] [hipe] [kernel-poll:false]

Eshell V5.6.5  (abort with ^G)
1> "hello world".
"hello world"
2>

With this in place I decided to try CouchDB. Again, here are the steps to get it running:


$ cd /downloads
$ tar -zxvf apache-couchdb-0.8.1-incubating.tar.gz 
$ cd apache-couchdb-0.8.1-incubating
$ less README 
$ sudo port install automake autoconf libtool help2man
$ sudo port install icu spidermonkey
$ ./configure 
$ make
$ sudo make install
$  sudo -i -u couchdb couchdb

I did not have to execute this line $ open /Applications/Installers/Xcode\ Tools/XcodeTools.mpkg. I assume it is because I had the full XCode install done beforehand. You can also see that you need to install the MacPorts tools. With those you will have to add a few binary packages to be able to build CouchDB. Especially the Unicode library ICU and the spidermonkey C based JavaScript library provided by the Mozilla organization. The last line starts up the database and directing your browser to http://localhost:5984/_utils/ allows you to see its internal UI. Now relax! ;)

If you look closely at the screen shot you will see an inconsistency to the notes above. I will describe this in more detail in a future post about getting the sample Twitter client for CouchDB running using CouchApp.

HBase vs. CouchDB in Berlin

2009-03-07T02:16:00.000-08:00

I had the pleasure of presenting our involvement with HBase at the 4th Berlin Hadoop Get Together. It was organized by Isabel Drost. Thanks again to Isabel for having me there, I thoroughly enjoyed it. First off, here are the slides:

HBase @ WorldLingo - Get more Information Technology

The second talk given was by Jan Lehnardt, a CouchDB team member. I am looking into Erlang for the last few months to see how we could use it for our own efforts. CouchDB is one of the projects you come across when reading articles about Erlang. So it was really great to have Jan present too.

At the end of both our talks it was great to see how the questions from the audience at times tried to compare the two. So is HBase better or worse than CouchDB. Of course, you cannot compare them directly. While they share common features (well, they store data, right?) they are made to solve different problems. CouchDB is offering a schema free storage with build in replication, which can even be used to create offline clients that sync their changes with another site when they have connectivity again. One of the features puzzling me most is the ability to use it to serve your own applications to the world. You create the pages and scripts you need and push it into the database using CouchApp. Since the database already has a built-in web server it can handle your applications requirements implicitly. Nice!

I asked Jan if he had concerns about scaling this, or if it wouldn't be better to use Apache or Nginx to serve the static content. His argument was that Erlang can handle many many more concurrent request than Apache can for example. I read up on Yaws and saw his point. So I guess it is a question then of memory and CPU requirements. The former is apparently another strength of CouchDB, which has proven to serve thousands of concurrent requests only needed about 10MB of RAM - how awesome is that?!?! I am not sure about CPU then - but take a gander that it is equally sane.

Another Erlang project I am interested in is Dynomite, a Erlang based Amazon Dynamo "clone" (or rather implementation). Talking to Cliff it seems it is as awesome leveraging the Erlang OTP abilities to create something that a normal Java developer and their JRE is just not used to.

And that brings me to HBase. I told the crowd in Berlin that as of version 0.18.0 HBase is ready for anyone to get started with - given they read the Wiki to set the file handles right and a few other bits in pieces.

Note: I was actually thinking about suggesting an improvement to the HBase team to have a command line check that can be invoked separately or is called when "start-hbase.sh" is called that checks a few of these common parameters and prints out warnings to the user. I know that the file handle count is printed out in the log files, but for a newbie this is a bit too deep down. What could be checked? First of the file handles being say 32K. The next thing is newer resource limits that were introduced with Hadoop for example that now need tweaking. An example is the "xciever" (sic) value. This again is documented in the Wiki, but who reads it, right? Another common issue is RAM. If the master knows the number of regions (or while it is scanning the META to determine it) it could warn if the JRE is not given enough memory. Sure, there are no hard boundaries, but better to see a Warning: Found x regions. Your configured memory for the JRE seems too low for the system to run stable.

Back to HBase. I also told the audience that as of HBase 0.19.0 the scanning was much improved speed wise and that I am happy where we are nowadays in terms of stability and speed. Sure, it could be faster for random reads so I may be able to drop my MemCached layer. And the team is working on that. So, here's hoping that we will see the best HBase ever in the upcoming version. I for myself am 100% sure that the HBase guys can deliver - they have done so in the past and will now as well. All I can say - give it a shot!

So, CouchDB is lean and mean while HBase is a resource hog from my experience. But it is also built to scale to Petabyte size data. With CouchDB, you would have to add sharding on top of it including all the issues that come with it, for example rebalancing, fail-over, recovery, adding more servers and so on. For me HBase is the system of choice - for this particular problem. That does not mean I could use CouchDB, or even Erlang for that matter, in a separate area. Until then I will keep my eyes very close in this exciting (though in case of Erlang not new!) technology. May the open-source projects are rule and live long and prosper!

Mini Local HBase Cluster

2009-02-21T18:13:00.000-08:00

I am trying to get a local setup going where I have everything I need on my PC - heck even on my MacBookPro. I use Eclipse and Java to develop, so that is easy. I also use Memcached and there is a nice MacPorts version for it available. But what I also need is a working Hadoop/HBase cluster!

At work we have a few of these, large and small, but they are either in production or simply to complex to use them for day to day testing. Especially when you try to debug a MapReduce job or code talking directly to HBase. I found that the excellent HBase team had already a class in place that is used to set up the JUnit tests they run. And the same goes for Hadoop. So I set out to extract the bare essentials if you will to create a tiny HBase cluster running on a tiny Hadoop distributed filesystem.

After a couple of issues that had to be resolved the below class is my "culmination" of sweet cluster goodness ;)


/* File:    MiniLocalHBase.java
 * Created: Feb 21, 2009
 * Author:  Lars George
 *
 * Copyright (c) 2009 larsgeorge.com
 */

package com.larsgeorge.hadoop.hbase;

import java.io.IOException;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.MiniHBaseCluster;
import org.apache.hadoop.hbase.util.FSUtils;
import org.apache.hadoop.hdfs.MiniDFSCluster;

/**
 * Starts a small local DFS and HBase cluster.
 *
 * @author Lars George
 */
public class MiniLocalHBase {

  static HBaseConfiguration conf = null;
  static MiniDFSCluster dfs = null;
  static MiniHBaseCluster hbase = null;
  
  /**
   * Main entry point to this class. 
   *
   * @param args  The command line arguments.
   */
  public static void main(String[] args) {
    try {
      int n = args.length > 0 && args[0] != null ? 
        Integer.parseInt(args[0]) : 4;
      conf = new HBaseConfiguration();
      dfs = new MiniDFSCluster(conf, 2, true, (String[]) null);
      // set file system to the mini dfs just started up
      FileSystem fs = dfs.getFileSystem();
      conf.set("fs.default.name", fs.getUri().toString());      
      Path parentdir = fs.getHomeDirectory();
      conf.set(HConstants.HBASE_DIR, parentdir.toString());
      fs.mkdirs(parentdir);
      FSUtils.setVersion(fs, parentdir);
      conf.set(HConstants.REGIONSERVER_ADDRESS, HConstants.DEFAULT_HOST + ":0");
      // disable UI or it clashes for more than one RegionServer
      conf.set("hbase.regionserver.info.port", "-1");
      hbase = new MiniHBaseCluster(conf, n);
      // add close hook
      Runtime.getRuntime().addShutdownHook(new Thread() {
        public void run() {
          hbase.shutdown();
          if (dfs != null) {
            try {
              FileSystem fs = dfs.getFileSystem();
              if (fs != null) fs.close();
            } catch (IOException e) {
              System.err.println("error closing file system: " + e);
            }
            try {
              dfs.shutdown();
            } catch (Exception e) { /*ignore*/ }
          }
        }
      } );
    } catch (Exception e) {
      e.printStackTrace();
    }
  } // main

} // MiniLocalHBase

The critical part for me was that if you wanted to be able to start more than one region server you have to disable the UI of each of these region servers or they will fail trying to bind the same info port, usually 60030.

I also added a small shutdown hook so that when you quit the process it will shut down nicely and keep the data in such a condition that you can restart the local again later on for further testing. Otherwise you may end up having to redo the file system - no biggie I guess, but hey why not? You can specify the number of RegionServer's being started on the command line. It defaults to 4 in my sample code above. Also, you do not need any hbase-site.xml or hadoop-site.xml to set anything else. All required settings are hardcoded to start the different servers in separate threads. You can of course add one and tweak further settings - just keep in mind that the ones hardcoded in the code cannot be reassigned by the external XML settings files. You would have to move those directly into the code.

To start this mini cluster you can either run this from within Eclipse for example, which makes it really easy since all the required libraries are in place, or you start it from the command line. This could work like so:


hadoop$ java -Xms512m -Xmx512m -cp bin:lib/hadoop-0.19.0-core.jar:lib/hadoop-0.19.0-test.jar:lib/hbase-0.19.0.jar:lib/hbase-0.19.0-test.jar:lib/commons-logging-1.0.4.jar:lib/jetty-5.1.4.jar:lib/servlet-api.jar:lib/jetty-ext/jasper-runtime.jar:lib/jetty-ext/jsp-api.jar:lib/jetty-ext/jasper-compiler.jar:lib/jetty-ext/commons-el.jar com.larsgeorge.hadoop.hbase.MiniLocalHBase

What I did is create a small project, have the class compile into the "bin" directory and threw all Hadoop and HBase libraries into the "lib" directory. This was only for the sake of keeping the command line short. I suggest you have the classpath set already or have it point to the original locations where you have untar'ed the respective packages.

Running it from within Eclipse let's you of course use the integrated debugging tools at hand. The next step is to follow through with what the test classes already have implemented and be able to start Map/Reduce jobs with the debugging enabled. Mind you though as the local cluster is not very powerful - even if you give it more memory than I did above. But fill it with a few hundred rows and use it to debug your code and once it runs fine, run it happily ever after on your production site.

All the credit goes to the Hadoop and HBase teams of course, I simply gathered their code from various places.

Apache fails on Semaphores

2009-02-05T10:21:00.001-08:00

In the last few years I had twice an issue with our Apache web servers where all of a sudden they would crash and not start again. While there are obvious reasons in case the configuration is screwed up there are also cases where you simply do not know why it would not restart. There is enough drive space, RAM, no other processes running locking the port (even checked with lsof).

All you get is an error message in the log saying:

[Fri May 21 15:34:22 2008] [crit] (28)No space left on device: mod_rewrite: could not create rewrite_log_lock
Configuration Failed

After some digging the issue was that all semaphores were used up and had to be deleted first. Here is a script I use to do that:

echo "Semaphores found: "
ipcs -s | awk '{ print $2 }' | wc -l
ipcs -s | awk '{ print $2 }' | xargs -n 1 ipcrm sem
echo "Semaphores found after removal: "
ipcs -s | awk '{ print $2 }' | wc -l

Sometimes you really wonder what else could go wrong.

String starts with a number in XSL

2009-02-04T03:48:00.001-08:00

I needed a way to test if a particular value in an XML file started with a letter based prefix. If not, then the value would start with a number and needed to be prefixed first before being output. While I found a great post how to remove leading zeros I could not find how to check if the first letter is of a particular type, for example a letter or a number. In Java you can do that easily like so (using BeanShell here):


bsh % String s1 = "1234";
bsh % String s2 = "A1234";
bsh % print(Character.isLetter(s1.charAt(0)));
false
bsh % print(Character.isLetter(s2.charAt(0)));
true

This is of course Unicode safe. With XSL though I could not find a similar feature but for my purposes it was sufficient to reverse the check and see if I had a Latin number first. Here is how:

<xsl:template match="/person/personnumber">
  <reference>
    <xsl:variable name="num">
      <xsl:value-of select="."/>
    </xsl:variable>
    <xsl:choose>
      <xsl:when test="contains('0123456789', substring($num, 1, 1))">
        <xsl:variable name="snum">
          <xsl:call-template name="removeLeadingZeros">
            <xsl:with-param name="originalString" select="$num"/>
          </xsl:call-template>
        </xsl:variable>
        <xsl:value-of select="concat('PE', $snum)"/>
      </xsl:when>
      <xsl:otherwise>
        <xsl:value-of select="."/>
      </xsl:otherwise>
    </xsl:choose>
  </reference>
</xsl:template>

Hadoop Scripts

2009-02-02T05:27:00.001-08:00

If you work with HBase and Hadoop in particular, you start off doing most things on the command line. After a while this is getting tedious and - in the end - becomes a nuisance. And error prone! While I wish there would be a an existing and established solution out there that helps managing a Hadoop cluster I find that there are few that you can use right now. Of the few that come to mind is the "Hadoop on Demand" (HOD) package residing in the contribution folder of the Hadoop releases. The other is ZooKeeper.

Interesting things are in the pipeline though, for example Simon from Yahoo!.

What I often need are small helpers that allow me to clean up behind me or which helps me deploy new servers. There are different solutions that usually involve some sort of combination of SSH and rsync. Tools I found and some of them even tried are SmartFrog, Puppet, and Tentakel.

Especially in the beginning you often find that you do not know these tools well enough or they do one thing - but not another. Of course, you can combine them and make that work somehow. I usually resort to a set of well known and proven scripts that I created over time to simplify working with a particular system. With Hadoop most of these scripts are run on the master and since it already is set up to use SSH to talk to all slaves it makes it easy to use the same mechanism.

The first one is to show all Java processes across the machines to see that they are up - or all shut down before attempting a new start:

#!/bin/bash
# $Revision: 1.0 $
#
# Shows all Java processes on the Hadoop cluster.
#
# Created 2008/01/07 by Lars George
#

servers="$(cat /usr/local/hadoop/conf/masters /usr/local/hadoop/conf/slaves)"

for srv in $servers; do
  echo "Sending command to $srv..."; 
  ssh $srv "ps aux | grep -v grep | grep java"
done

echo "done."

The next one is a poor-man's deployment thingamajig. It helps copying a new release across the machines and setting up the symbolic link I use for the current version in production. Of course this all varies with your setup.

#!/bin/bash
# $Revision: 1.0 $
#
# Rsync's Hadoop files across all slaves. Must run on namenode.
#
# Created 2008/01/03 by Lars George
#

if [ "$#" != "2" ]; then
  echo "usage: $(basename $0) <dir-name> <ln-name>"
  echo "  example: $(basename $0) hbase-0.1 hbase"
  exit 1
fi

for srv in $(cat /usr/local/hadoop/conf/slaves); do
  echo "Sending command to $srv..."; 
  rsync -vaz --exclude='logs/*' /usr/local/$1 $srv:/usr/local/
  ssh $srv "rm -fR /usr/local/$2 ; ln -s /usr/local/$1 /usr/local/$2"
done

echo "done."

I basically download a new version on the master (or build one) and issue a

$ rsyncnewhadoop /usr/local/hadoop-0.19.0 hadoop

It copies the directory across and changes the "/usr/local/hadoop" symbolic link to point to the this new release.

Another helper I use quite often is to diff an existing and a new version before I actually copy them across the cluster. It can be used like so:

$ diffnewversion /usr/local/hbase-0.19.0 /usr/local/hadoop-0.19.0

Again I assume that the current version is symlinked as explained above. Otherwise you would have to make adjustements obviously.

#!/bin/bash
#
# Diff's the configuration files between the current symlinked versions and the given one.
#
# Created 2009/01/23 by Lars George
#

if [[ $# == 0 ]]; then
  echo "usage: $(basename $0) <new_dir> [<new_dir>]"
  exit 1;
fi

DIRS="conf bin"

for path in $*; do
  if [[ "$1" == *hadoop* ]]; then
    kind="hadoop"
  else
    kind="hbase"
  fi
  for dir in $DIRS; do  
    echo
    echo "Comparing $kind $dir directory..."
    echo
    for f in /usr/local/$kind/$dir/*; do 
      echo
      echo
      echo "Checking $(basename $f)" 
      diff -w $f $1/$dir/$(basename $f)
      if [[ $? == 0 ]]; then
        echo "Files are the same..."
      fi
      echo 
      echo "================================================================"
    done 
  done
  shift 1
done

echo "done."

The last one I am posting here helps removing the Distributed File System (DFS) after for example a complete corruption (I didn't say they happen) or when you want to have a clean start.

Note: It assumes that the data is stored under "/data1/hadoop" and "/data2/hadoop" - that is where I have my data. If yours is different then adjust the path or - if you like - grep/awk the hadoop-site.xml and parse the paths out of the "dfs.name.dir" and "dfs.data.dir" respectively.

#!/bin/bash
# $Revision: 1.0 $
#
# Deletes all files and directories pertaining to the Hadoop DFS.
#
# Created 2008/12/12 by Lars George
#

servers="$(cat /usr/local/hadoop/conf/masters /usr/local/hadoop/conf/slaves)"
# optionally allow single server use
if [[ $# > 0 ]]; then 
  servers="$*"
fi
first="$(echo $servers | head -n 1 | awk -F. '{ print $1 }')"
dirs="/tmp/hbase* /tmp/hsperfdata* /tmp/task* /tmp/Jetty* /data1/hadoop/* /data2/hadoop/*"

echo "IMPORTANT: Are you sure you want to delete the DFS starting with $first?"
echo "Type \"yes\" to continue:"
read yes
if [ "$yes" == "yes" ]; then
  for srv in $servers; do
    echo "Sending command to $srv..."; 
    for dir in $dirs; do
      pa=$(dirname $dir)
      fn=$(basename $dir)
      echo "removing $pa/$fn...";
      ssh $srv "find $pa -name \"$fn\" -type f -delete ; rm -fR $pa/$fn" 
    done
  done
else 
  echo "aborted."
fi

echo "done."

I have a few others that for example let me kill runaway Java processes, sync only config changes across the cluster machines, starts and stops safely, and so on. I won't post them here as they are pretty trivial like the ones above or do not differ much. Let me know if you have similar scripts or better ones!