HBaseWD: Avoid RegionServer Hotspotting Despite Sequential Keys

In HBase world, RegionServer hotspotting is a common problem.  We can describe this problem with a single sentence: while writing records with sequential row keys allows the most efficient reading of data range given the start and stop keys, it causes undesirable RegionServer hotspotting at write time. In this 2-part post series we’ll discuss the problem and show you how to avoid this infamous problem.

Problem Description

Records in HBase are sorted lexicographically by the row key. This allows fast access to an individual record by its key and fast fetching of a range of data given start and stop keys. There are common cases where you would think row keys forming a natural sequence at write time would be a good choice because of  types of queries that will fetch the data later. For example, we may want to associate each record with a timestamp so that later we can fetch records from a particular time range.  Examples of such keys are:

  • time-based format: Long.MAX_VALUE – new Date().getTime()
  • increasing/decreasing sequence: ”001”, ”002”, ”003”,… or ”499”, ”498”, ”497”, …

But writing records with such naive keys will cause hotspotting because of how HBase writes data to its Regions.

RegionServer Hotspotting

When records with sequential keys are being written to HBase all writes hit one Region.  This would not be a problem if a Region was served by multiple RegionServers, but that is not the case – each Region lives on just one RegionServer.  Each Region has a pre-defined maximal size, so after a Region reaches that size it is split in two smaller Regions.  Following that, one of these new Regions takes all new records and then this Region and the RegionServer that serves it becomes the new hotspot victim.  Obviously, this uneven write load distribution is highly undesirable because it limits the write throughput to the capacity of a single server instead of making use of multiple/all nodes in the HBase cluster. The uneven load distribution can be seen in Figure 1. (chart courtesy of SPM for HBase):

HBase RegionServer Hotspotting

Figure 1. HBase RegionServer hotspotting

We can see that while one server was sweating trying to keep up with writes, others were “resting”. You can find some more information about this problem in HBase Reference Guide.

Solution Approach

So how do we solve this problem?  The cases discussed here assume that we don’t have all data we want to write to HBase all at once, but rather that the data are arriving continuously. In case of bulk import of data into HBase the best solutions, including those that avoid hotspotting, are described in bulk load section of HBase documentation.  However, if you are like us at Sematext, and many organizations nowadays are, the data keeps streaming in and needs processing and storing. The simplest way to avoid single RegionServer hotspotting in case of continuously arriving data would be to simply distribute writes over multiple Regions by using random row keys. Unfortunately, this would compromise ability to do fast range scans using start and stop keys. But that is not the only solution.  The following simple approach solves the hotspotting issue while at the same time preserving the ability to fetch data by start and stop key.  This solution, mentioned multiple times on HBase mailing lists and elsewhere is to salt row keys with a prefix.  For example, consider constructing the row key using this:

new_row_key = (++index % BUCKETS_NUMBER) + original_key

For the visual types among us, that may result in keys looking as shown in Figure 2.

HBase Row Key Prefix Salting

Figure 2. HBase row key prefix salting

Here we have:

  • index is the numeric (or any sequential) part of the specific record/row ID that we later want to use for record fetching (e.g. 1, 2, 3 ….)
  • BUCKETS_NUMBER is the number of “buckets” we want our new row keys to be spread across. As records are written, each bucket preserves the sequential notion of original records’ IDs
  • original_key is the original key of the record we want to write
  • new_row_key is the actual key that will be used when writing a new record (i.e. “distributed key” or “prefixed key”). Later in the post the “distributed records” term is used for records which were written with this “distributed key”.

Thus, new records will be split into multiple buckets, each (hopefully) ending up in a different Region in the HBase cluster. New row keys of bucketed records will no longer be in one sequence, but records in each bucket will preserve their original sequence. Of course, if you start writing into an empty HTable, you’ll have to wait some time (depending on the volume and velocity of incoming data, compression, and maximal Region size) before you have several Regions for a table. Hint: use pre-splitting feature for the new table to avoid the wait time.  Once writes using the above approach kick in and start writing to multiple Regions your “slaves load” chart should look better.

HBase RegionServer evenly distributed write load

Figure 3. HBase RegionServer evenly distributed write load

Scan

Since data is placed in multiple buckets during writes, we have to read from all of those buckets when doing scans based on “original” start and stop keys and merge data so that it preserves the “sorted” attribute. That means BUCKETS_NUMBER more Scans and this can affect performance. Luckily, these scans can be run in parallel and performance should not degrade or might even improve — compare the situation when you read 100K sequential records from one Region (and thus one RegionServer) with reading 10K records from 10 Regions and 10 RegionServers in parallel!

Get/Delete

To get or delete a single record by original key may need to perform 1 or up to BUCKETS_NUMBER Get operations depending on the logic we used for prefix generation. E.g. when using “static” hash as prefix, given the original key we may precisely identify the prefixed key. In case we used random prefix we will have to perform Get for each of the possible buckets. The same goes for Delete operations.

MapReduce Input

Since we still want to benefit from data locality, the implementation of feeding “distributed” data to a MapReduce job will likely break the order in which data comes to mappers. This is at least true for the current HBaseWD implementation (see below). Each map task will process data for a particular bucket. Of course, records will be in same order based on original keys within a bucket. However, since two records which were meant to be “near each other” based on their original key may have fallen into different buckets, the will be fed to different map tasks. Thus, if the mapper assumes records come in the strict/original sequence, we will be hurt, since the order will be preserved only within each bucket, but not globally.

Increased Number of Map Tasks

When using data (written using the suggested approach) as a MapReduce input (with start and/or stop key provided) the splits number will likely to be increased (depends on the implementation). For current HBaseWD implementation you’ll get BUCKETS_NUMBER times more splits compared to “regular” MapReduce with same parameters.  This is due to the same logic for data selection as with simple Scan operation, as described above. As the result, MapReduce jobs will have BUCKETS_NUMBER times more map tasks. This should not decrease performance if BUCKETS_NUMBER is reasonably not too high (when MR job initialization & cleanup work takes more time than processing itself). Moreover, in many use-cases having many more mappers helps improve performance. Many users reported more mappers having a positive impact given that standard HTable input based MapReduce job usually has too few map tasks (one per Region) which cannot be changed without extra coding.

Another strong signal the suggested approach and its implementation could help is if in your application, in addition to writing records with sequential keys, the application also continuously processes newly written data delta using MapReduce . In such use-cases when data is written sequentially (not using any artificial distribution) and is being processed relatively frequently, the delta to be processed resides in just a few Regions (or perhaps in even just one Region if the write load is not high, if maximal Region size is high, and processing batches are very frequent).

Solution Implementation: HBaseWD

We implemented the solution described above and open-sourced it as a small HBaseWD project. We say small because HBaseWD is really self-contained and really simple to integrate into an existing code due to support for native HBase client API (see examples below). HBaseWD project was first presented at BerlinBuzzwords 2011(video) and is currently used in a number of production systems.

Configuring Distribution

Simple Even Distribution

Distributing records with sequential keys to be distributed in up to Byte.MAX_VALUE buckets (single byte is added in front of a key):

byte bucketsCount = (byte) 32; // distributing into 32 buckets
RowKeyDistributor keyDistributor =  new RowKeyDistributorByOneBytePrefix(bucketsCount);
Put put = new Put(keyDistributor.getDistributedKey(originalKey));
... // add values
hTable.put(put);

Hash-Based Distribution

Another useful RowKeyDistributor is RowKeyDistributorByHashPrefix. Please see example below. It creates the “distributed key” based on original key value so that later when you have original key and want to update the record you can calculate distributed key without having to call HBase (too see what bucket it is in). Or, you can perform a single Get operation when original key is known (instead of reading from all buckets).

AbstractRowKeyDistributor keyDistributor =
     new RowKeyDistributorByHashPrefix(
            new RowKeyDistributorByHashPrefix.OneByteSimpleHash(15));

You can use your own hashing logic here by implementing this simple interface:

public static interface Hasher extends Parametrizable {
  byte[] getHashPrefix(byte[] originalKey);
  byte[][] getAllPossiblePrefixes();
}

Custom Distribution Logic

HBaseWD is designed to be flexible especially when it comes to supporting custom row key distribution approaches. In addition to the above mentioned ability to implement custom hashing logic to be used with RowKeyDistributorByHashPrefix, one can define custom row key distribution logic by extending AbstractRowKeyDistributor abstract class whose interface is super simple:

public abstract class AbstractRowKeyDistributor implements Parametrizable {
  public abstract byte[] getDistributedKey(byte[] originalKey);
  public abstract byte[] getOriginalKey(byte[] adjustedKey);
  public abstract byte[][] getAllDistributedKeys(byte[] originalKey);
  ... // some utility methods
}

Common Operations

Scan

Performing a range scan over data:

Scan scan = new Scan(startKey, stopKey);
ResultScanner rs = DistributedScanner.create(hTable, scan, keyDistributor);
for (Result current : rs) {
  ...
}

Configuring MapReduce Job

Performing MapReduce job over the data chunk specified by Scan:

Configuration conf = HBaseConfiguration.create();
Job job = new Job(conf, "testMapreduceJob");
Scan scan = new Scan(startKey, stopKey);
TableMapReduceUtil.initTableMapperJob("table", scan,
RowCounterMapper.class, ImmutableBytesWritable.class, Result.class, job);
// Substituting standard TableInputFormat which was set in
// TableMapReduceUtil.initTableMapperJob(...)
job.setInputFormatClass(WdTableInputFormat.class);
keyDistributor.addInfo(job.getConfiguration());

What’s Next?

In the next post we’ll cover:

  • Integration into already running production systems
  • Changing distribution logic in running systems
  • Other “advanced topics”
If you’ve read this far you must be interested in HBase.  And since we (@sematext) are interested in people with interest in HBase, here are some open positions at Sematext, some nice advantages we offer and “problems” we are into that you may want to check out.
Oh, and all HBase performance charts you see in this post are from our SPM service which uses HBase and, you guessed it, HBaseWD too! ;)

22 Responses to HBaseWD: Avoid RegionServer Hotspotting Despite Sequential Keys

  1. Jason says:

    what’s the difference between using buckets vs using a random int between 0-9? Prefix the sequential numeric keys with a 0-9 random number may also work and the code is much simpler?

    • Lambda says:

      if your hbase cluster is composed of 20 tablet servers and you have your original keys prefixed with random integers between 0 to 9, can you guarantee that your prefixed keys are distribued uniformlly among the 20 tablet servers?

  2. Jason says:

    If we don’t need a range scan option and don’t need to put in any Map-reduce job. Then using a random number prefix may also work (avoid hot-spots) while keeping the solution simple?

    • Alex Baranau says:

      Right, if you don’t care about reading the data, then things becoming very simple.

      Actually what HBaseWD does is exactly what you are describing: prefixing keys with some known values. It’s just that HBaseWD also implements those goodies (range scans, feeding data in MR) for you.

  3. Bing says:

    I think your hbase monitor system makes sense, and can you tell me how you build your hbase monitor system. I think “Figure 3. HBase RegionServer evenly distributed write load” is very useful.

    • sematext says:

      @Bing It’s little bit difficult to explain how we built our HBase monitoring system in a blog post comment, but if you have some specific question, we can try answering it.

      • Bing says:

        In our system, We add some basic monitor on hbase, such as Regionserver liveness, the change of Region, and HBase client,however, there is nothing to do with hbase real-time read/write performance. So if you can share some documents about it, it will be very useful.

  4. rahul says:

    First of all Thank You. Your post was very helpful. Does Hbase ensure that all records in a bucket are placed on a single region/regionserver or is HBase completely unaware of the concept of a bucket.

    • Alex Baranau says:

      You are welcome!

      HBase is completely unaware of the concept of a bucket. That is OK: you don’t need the bucket to be stored on one RS. Moreover, your single buckets will likely not fit into one region, which means that data may be spread across RSs (along with those regions).

      • rahul says:

        Thank you. That makes a lot of sense when we try to think through what HBase does when it starts with an empty table to the point where we have atleast 3-4 regions.

        • Alex Baranau says:

          yup. Just don’t forget to look at pre-splitting the table (http://hbase.apache.org/book/perf.writing.html) so that you don’t *actually* start with one region ;)

          • rahul says:

            Thank you. A couple of things are still poking me. Memstore in HBase performs the sorting and the flushes the contents to a HFile. If the Region data is sorted by rowkey, then we should be able to insert the record in the middle of a file(HDFS does not support this) and not just append to it. The word append is also confusing since we cannot open a closed file and start appending to it, we can only open it as long as the stream is open. Correct me if I am wrong. Thanks in advance for taking time to read this.

  5. Alex Baranau says:

    @rahul: RE your last comment. Every time flush happens HBase creates *new* file in HDFS. And later these files are merged (with compation) or data from them is merged at read time (on the fly). No need to change existing files in HDFS. This post maaay be useful to read: http://blog.cloudera.com/blog/2012/06/hbase-io-hfile-input-output/

  6. jtaylorsfdc says:

    Coincidently, a post on the Phoenix mailing list led me back to you again, Alex. Nice work! Another good candidate to incorporate into Phoenix. We already have an issue for this one: https://github.com/forcedotcom/phoenix/issues/74. That’s two you can help us with now. :-)
    But seriously, let me know if you’d like to get involved and I’ll give you a more detailed idea and code pointers for integrating your nice work.

    Regards,
    James

  7. haqyunus says:

    This has been a blessing. Thanks a lot. I have a question about the post mentioned in the ‘What’s Next’ section? Was that ever written? A link would be great as I was unable to find it.

    • Alex Baranau says:

      Not yet unfortunately. I got new job (and a kid on top of that) and (as you can imagine) had to focus on it and “disappear” for some time. Really sorry for that. I still plan to write the post, but can’t promise any ETA unfortunately.

  8. sauragar says:

    Wonderful article. Thanks for sharing. One question. You had mentioned that “Luckily, these scans can be run in parallel and performance should not degrade or might even improve”. But in the code, The scan calls are getting invoked sequentially. How is the parallelism achieved here? Thanks.

Leave a Reply

Fill in your details below or click an icon to log in:

WordPress.com Logo

You are commenting using your WordPress.com account. Log Out / Change )

Twitter picture

You are commenting using your Twitter account. Log Out / Change )

Facebook photo

You are commenting using your Facebook account. Log Out / Change )

Google+ photo

You are commenting using your Google+ account. Log Out / Change )

Connecting to %s

Follow

Get every new post delivered to your Inbox.

Join 1,695 other followers

%d bloggers like this: