The first concept to explain in more detail is the logical layout of a table, compared to on-disk storage. HBase’s main unit of separation within a table is the column family—not the actual columns as expected from a column-oriented database in their traditional sense. Figure 9-1 shows the fact that, although you store cells in a table format logically, in reality these rows are stored as linear sets of the actual cells, which in turn contain all the vital information inside them.

The top-left part of the figure shows the logical layout of your data—you have rows and columns. The columns are the typical HBase combination of a column family name and a column qualifier, forming the column key. The rows also have a row key so that you can address all columns in one logical row.

Figure 9-1. Rows stored as linear sets of actual cells, which contain all the vital information

The top-right hand side shows how the logical layout is folded into the actual physical storage layout. The cells of each row are stored one after the other, in a separate storage file per column family. In other words, on disk you will have all cells of one family in a StoreFile, and all cells of another in a different file.

Since HBase is not storing any unset cells (also referred to as NULL values by RDBMSes) from the table, the on-disk file only contains the data that has been explicitly set. It therefore has to also store the row key and column key with every cell so that it can retain this vital piece of information.

In addition, multiple versions of the same cell are stored as separate, consecutive cells, adding the required timestamp of when the cell was stored. The cells are sorted in descending order by that timestamp so that a reader of the data will see the newest value first—which is the canonical access pattern for the data.

The entire cell, with the added structural information, is called KeyValue in HBase terms. It has not just the column and actual value, but also the row key and timestamp, stored for every cell for which you have set a value. The KeyValues are sorted by row key first, and then by column key in case you have more than one cell per row in one column family.

The lower-right part of the figure shows the resultant layout of the logical table inside the physical storage files. The HBase API has various means of querying the stored data, with decreasing granularity from left to right: you can select rows by row keys and effectively reduce the amount of data that needs to be scanned when looking for a specific row, or a range of rows. Specifying the column family as part of the query can eliminate the need to search the separate storage files. If you only need the data of one family, it is highly recommended that you specify the family for your read operation.

Although the timestamp—or version—of a cell is farther to the right, it is another important selection criterion. The store files retain the timestamp range for all stored cells, so if you are asking for a cell that was changed in the past two hours, but a particular store file only has data that is four or more hours old it can be skipped completely. See also Read Path for details.

The next level of query granularity is the column qualifier. You can employ exact column lookups when reading data, or define filters that can include or exclude the columns you need to access. But as you will have to look at each KeyValue to check if it should be included, there is only a minor performance gain.

The value remains the last, and broadest, selection criterion, equaling the column qualifier’s effectiveness: you need to look at each cell to determine if it matches the read parameters. You can only use a filter to specify a matching rule, making it the least efficient query option. Figure 9-2 summarizes the effects of using the KeyValue fields.

Figure 9-2. Retrieval performance decreasing from left to right

The crucial part of Figure 9-1 shows is the shift in the lower-lefthand side. Since the effectiveness of selection criteria greatly diminishes from left to right for a KeyValue, you can move all, or partial, details of the value into a more significant place—without changing how much data is stored.

At this time, you may be asking yourself where and how you should store your data. The two choices are tall-narrow and flat-wide. The former is a table with few columns but many rows, while the latter has fewer rows but many columns. Given the explained query granularity of the KeyValue information, it seems to be advisable to store parts of the cell’s data—especially the parts needed to query it—in the row key, as it has the highest cardinality.

In addition, HBase can only split at row boundaries, which also enforces the recommendation to go with tall-narrow tables. Imagine you have all emails of a user in a single row. This will work for the majority of users, but there will be outliers that will have magnitudes of emails more in their inbox—so many, in fact, that a single row could outgrow the maximum file/region size and work against the region split facility.

The better approach would be to store each email of a user in a separate row, where the row key is a combination of the user ID and the message ID. Looking at Figure 9-1 you can see that, on disk, this makes no difference: if the message ID is in the column qualifier, or in the row key, each cell still contains a single email message. Here is the flat-wide layout on disk, including some examples:

<userId> : <colfam> : <messageId> : <timestamp> : <email-message>

12345 : data : 5fc38314-e290-ae5da5fc375d : 1307097848 : "Hi Lars, ..."
12345 : data : 725aae5f-d72e-f90f3f070419 : 1307099848 : "Welcome, and ..."
12345 : data : cc6775b3-f249-c6dd2b1a7467 : 1307101848 : "To Whom It ..."
12345 : data : dcbee495-6d5e-6ed48124632c : 1307103848 : "Hi, how are ..."

The same information stored as a tall-narrow table has virtually the same footprint when stored on disk:

<userId>-<messageId> : <colfam> : <qualifier> : <timestamp> : <email-message>

12345-5fc38314-e290-ae5da5fc375d : data : : 1307097848 : "Hi Lars, ..."
12345-725aae5f-d72e-f90f3f070419 : data : : 1307099848 : "Welcome, and ..."
12345-cc6775b3-f249-c6dd2b1a7467 : data : : 1307101848 : "To Whom It ..."
12345-dcbee495-6d5e-6ed48124632c : data : : 1307103848 : "Hi, how are ..."

This layout makes use of the empty qualifier (see Column Families). The message ID is simply moved to the left, making it more significant when querying the data, but also transforming each email into a separate logical row. This results in a table that is easily splittable, with the additional benefit of having a more fine-grained query granularity.

The scan functionality of HBase, and the HTable-based client API, offers the second crucial part for transforming a table into a tall-narrow one, without losing query granularity: partial key scans.

In the preceding example, you have a separate row for each message, across all users. Before you had one row per user, so a particular inbox was a single row and could be accessed as a whole. Each column was an email message of the users’ inbox. The exact row key would be used to match the user ID when loading the data.

With the tall-narrow layout an arbitrary message ID is now postfixed to the user ID in each row key. If you do not have an exact combination of these two IDs you cannot retrieve a particular message. The way to get around this complication is to use partial key scans: you can specify a start and end key that is set to the exact user ID only, with the stop key set to userId + 1.

The start key of a scan is inclusive, while the stop key is exclusive. Setting the start key to the user ID triggers the internal lexicographic comparison mechanism of the scan to find the exact row key, or the one sorting just after it. Since the table does not have an exact match for the user ID, it positions the scan at the next row, which is:

<userId>-<lowest-messageId>

In other words, it is the row key with the lowest (in terms of sorting) user ID and message ID combination. The scan will then iterate over all the messages of a user and you can parse the row key to extract the message ID.

The partial key scan mechanism is quite powerful, as you can use it as a lefthand index, with each added field adding to its cardinality. Consider the following row key structure:

<userId>-<date>-<messageId>-<attachmentId>

You can, with increasing precision, construct a start and stop key for the scan that selects the required rows. Usually you only create the start key and set the stop key to the same value as the start key, while increasing the least significant byte of its first field by one. For the preceding inbox example, the start key could be 12345, and the stop key 123456.

Table 9-1 shows the possible start keys and what they translate into.

These composite row keys are similar to what RDBMSes offer, yet you can control the sort order for each field separately. For example, you could do a bitwise inversion of the date expressed as a long value (the Linux epoch). This would then sort the rows in descending order by date. Another approach is to compute the following:

Long.MAX_VALUE - <date-as-long>

This will reverse the dates and guarantee that the sorting order of the date field is descending.

In the preceding example, you have the date as the second field in the composite index for the row key. This is only one way to express such a combination. If you were to never query by date, you would want to drop the date from the key—and/or possibly use another, more suitable, dimension instead.

Using the partial key scan approach, it is possible to iterate over subsets of rows. The principle is the same: you have to specify an appropriate start and stop key to limit the overall number of rows scanned. Then you take an offset and limit parameter, applying them to the rows on the client side.

The steps are the following:

Applying this to the inbox example, it is possible to paginate through all of the emails of a user. Assuming an average user has a few hundred emails in his inbox, it is quite common for a web-based email client to show only the first, for example, 50 emails. The remainder of the emails are then accessed by clicking the Next button to load the next page.

The client would set the start row to the user ID, and the stop row to the user ID + 1. The remainder of the process would follow the approach we just discussed, so for the first page, where the offset is zero, you can read the next 50 emails. When the user clicks the Next button, you would set the offset to 50, therefore skipping those first 50 rows, returning row 51 to 100, and so on.

This approach works well for a low number of pages. If you were to page through thousands of pages, a different approach would be required. You could add a sequential ID into the row key to directly position the start key at the right offset. Or you could use the date field of the key—if you are using one—to remember the date of the last displayed item and add the date to the start key, but probably dropping the hour part of it. If you were using epochs, you could compute the value for midnight of the last seen date. That way you can rescan that entire day and make a more knowledgeable decision regarding what to return.

There are many ways to design the row key to allow for efficient selection of subranges and enable pagination through records, such as the emails in the user inbox example. Using the composite row key with the user ID and date gives you a natural order, displaying the newest messages first, sorting them in descending order by date. But what if you also want to offer sorting by different fields so that the user can switch at will? One way to do this is discussed in Secondary Indexes.

When dealing with stream processing of events, the most common use case is time series data. Such data could be coming from a sensor in a power grid, a stock exchange, or a monitoring system for computer systems. Its salient feature is that its row key represents the event time. This imposes a problem with the way HBase is arranging its rows: they are all stored sorted in a distinct range, namely regions with specific start and stop keys.

The sequential, monotonously increasing nature of time series data causes all incoming data to be written to the same region. And since this region is hosted by a single server, all the updates will only tax this one machine. This can cause regions to really run hot with the number of accesses, and in the process slow down the perceived overall performance of the cluster, because inserting data is now bound to the performance of a single machine.

It is easy to overcome this problem by ensuring that data is spread over all region servers instead. This can be done, for example, by prefixing the row key with a nonsequential prefix. Common choices include:

Summarizing the various approaches, you can see that it is not trivial to find the right balance between optimizing for read and write performance. It depends on your access pattern, which ultimately drives the decision on how to structure your row keys. Figure 9-3 shows the various solutions and how they affect sequential read and write performance.

Figure 9-3. Finding the right balance between sequential read and write performance

Using the salted or promoted field keys can strike a good balance of distribution for write performance, and sequential subsets of keys for read performance. If you are only doing random reads, it makes most sense to use random keys: this will avoid creating region hot-spots.

In our preceding discussion, the time series data dealt with inserting new events as separate rows. However, you can also store related, time-ordered data: using the columns of a table. Since all of the columns are sorted per column family, you can treat this sorting as a replacement for a secondary index, as available in RDBMSes. Multiple secondary indexes can be emulated by using multiple column families—although that is not the recommended way of designing a schema. But for a small number of indexes, this might be what you need.

Consider the earlier example of the user inbox, which stores all of the emails of a user in a single row. Since you want to display the emails in the order they were received, but, for example, also sorted by subject, you can make use of column-based sorting to achieve the different views of the user inbox.

The first decision to make concerns what the primary sorting order is, in other words, how the majority of users have set the view of their inbox. Assuming they have set the view in descending order by date, you can use the same approach mentioned earlier, which reverses the timestamp of the email, effectively sorting all of them in descending order by time:

Long.MAX_VALUE - <date-as-long>

The email itself is stored in the main column family, while the sort indexes are in separate column families. You can extract the subject from the email address and add it to the column key to build the secondary sorting order. If you need descending sorting as well, you would need another family.

To circumvent the proliferation of column families, you can alternatively store all secondary indexes in a single column family that is separate from the main column family. Once again, you would make use of implicit sorting by prefixing the values with an index ID—for example, idx-subject-desc, idx-to-asc, and so on. Next, you would have to attach the actual sort value. The actual value of the cell is the key of the main index, which also stores the message. This also implies that you need to either load the message details from the main table, display only the information stored in the secondary index, or store the display details redundantly in the index, avoiding the random lookup on the main information source. Recall that denormalization is quite common in HBase to reduce the required read operations in favor of vastly improved user-facing responsiveness.

Putting the aforementioned schema into action might result in something like this:

12345 : data : 5fc38314-e290-ae5da5fc375d : 1307097848 : "Hi Lars, ..."
12345 : data : 725aae5f-d72e-f90f3f070419 : 1307099848 : "Welcome, and ..."
12345 : data : cc6775b3-f249-c6dd2b1a7467 : 1307101848 : "To Whom It ..."
12345 : data : dcbee495-6d5e-6ed48124632c : 1307103848 : "Hi, how are ..."
...
12345 : index : [email protected] : 1307099848 : 725aae5f-d72e...
12345 : index : [email protected] : 1307103848 : dcbee495-6d5e...
12345 : index : [email protected] : 1307097848 : 5fc38314-e290...
12345 : index : [email protected] : 1307101848 : cc6775b3-f249...
...
12345 : index : idx-subject-desc-xa8x90x8dx93x9bxde : 
  1307103848 : dcbee495-6d5e-6ed48124632c
12345 : index : idx-subject-desc-xb7x9ax93x93x90xd3 : 
  1307099848 : 725aae5f-d72e-f90f3f070419
...

In the preceding code, one index (idx-from-asc) is sorting the emails in ascending order by from address, and another (idx-subject-desc) in descending order by subject. The subject itself is not readable anymore as it was bit-inversed to achieve the descending sorting order. For example:

% String s = "Hello,";
% for (int i = 0; i < s.length(); i++) { 
  print(Integer.toString(s.charAt(i) ^ 0xFF, 16)); 
}
b7 9a 93 93 90 d3

All of the index values are stored in the column family index, using the prefixes mentioned earlier. A client application can read the entire column family and cache the content to let the user quickly switch the sorting order. Or, if the number of values is large, the client can read the first 10 columns starting with idx-subject-desc to show the first 10 email messages sorted in ascending order by the email subject lines. Using a scan with intra-row batching (see Caching Versus Batching) enables you to efficiently paginate through the subindexes. Another option is the ColumnPaginationFilter, combined with the ColumnPrefixFilter to iterate over an index page by page.

I pointed out before that you should ensure that the clock on your servers is synchronized. Otherwise, when you store data in multiple rows across different servers, using the implicit timestamps, you may end up with completely different time settings.

For example, say you use the HBase URL Shortener and store three new shortened URLs for an existing user. All of the keys are considered fully distributed, so all three of the new rows end up on a different region server. Further, assuming that these servers are all one hour apart, if you were to scan from the client side to get the list of new shortened URLs within the past hour, you would miss a few, as they have been saved with a timestamp that is more than an hour different from what the client considers current.

This can be avoided by setting an agreed, or shared, timestamp when storing these values. The put operation allows you to set a client-side timestamp that is used instead, therefore overriding the server time. Obviously, the better approach is to rely on the servers doing this work for you, but you might be required to use this approach in some circumstances.^[107]

Another issue with servers not being aligned by time is exposed by region splits. Assume you have saved a value on a server that is one hour ahead all other servers in the cluster, using the implicit timestamp of the server. Ten minutes later the region is split and the half with your update is moved to another server. Five minutes later you are inserting a new value for the same column, again using the automatic server time. The new value is now considered older than the initial one, because the first version has a timestamp one hour ahead of the current server’s time. If you do a standard get call to retrieve the newest version of the value, you will get the one that was stored first.

Once you have all the servers synchronized, there are a few more interesting side effects you should know about. First, it is possible—for a specific time—to make versions of a column reappear. This happens when you store more versions than are configured at the column family level. The default is to keep the last three versions of a cell, or value.

If you insert a new value 10 times into the same column, and request a complete list of all versions retained, using the setMaxVersions() call of the Get class, you will only ever receive up to what is configured in the table schema, that is, the last three versions by default.

But what would happen when you explicitly delete the last two versions? Example 9-1 demonstrates this.

    for (int count = 1; count <= 6; count++) { 
      Put put = new Put(ROW1);
      put.add(COLFAM1, QUAL1, count, Bytes.toBytes("val-" + count)); 
      table.put(put);
    }

    Delete delete = new Delete(ROW1); 
    delete.deleteColumn(COLFAM1, QUAL1, 5);
    delete.deleteColumn(COLFAM1, QUAL1, 6);
    table.delete(delete);

When you run the example, you should see the following output:

After put calls...
KV: row1/colfam1:qual1/6/Put/vlen=5, Value: val-6
KV: row1/colfam1:qual1/5/Put/vlen=5, Value: val-5
KV: row1/colfam1:qual1/4/Put/vlen=5, Value: val-4
After delete call...
KV: row1/colfam1:qual1/4/Put/vlen=5, Value: val-4
KV: row1/colfam1:qual1/3/Put/vlen=5, Value: val-3
KV: row1/colfam1:qual1/2/Put/vlen=5, Value: val-2

An interesting observation is that you have resurrected versions 2 and 3! This is caused by the fact that the servers delay the housekeeping to occur at well-defined times. The older versions of the column are still kept, so deleting newer versions makes the older versions come back.

This is only possible until a major compaction has been performed, after which the older versions are removed forever, using the predicate delete based on the configured maximum versions to retain.

After put calls...
KV: row1/colfam1:qual1/6/Put/vlen=5, Value: val-6
KV: row1/colfam1:qual1/5/Put/vlen=5, Value: val-5
KV: row1/colfam1:qual1/4/Put/vlen=5, Value: val-4
After delete call...
KV: row1/colfam1:qual1/4/Put/vlen=5, Value: val-4

Finally, when dealing with timestamps, there is another issue to watch out for: delete markers. This refers to the fact that, in HBase, a delete is actually adding a tombstone marker into the store that has a specific timestamp. Based on that, it masks out versions that are either a direct match, or, in the case of a column delete marker, anything that is older than the given timestamp. Example 9-2 shows this using the shell.

hbase(main):001:0> create 'testtable', 'colfam1'
0 row(s) in 1.1100 seconds

hbase(main):002:0> Time.now.to_i
=> 1308900346

hbase(main):003:0> put 'testtable', 'row1', 'colfam1:qual1', 'val1' 
0 row(s) in 0.0290 seconds

hbase(main):004:0> scan 'testtable'
ROW   COLUMN+CELL
 row1 column=colfam1:qual1, timestamp=1308900355026, value=val1
1 row(s) in 0.0360 seconds

hbase(main):005:0> delete 'testtable', 'row1', 'colfam1:qual1' 
0 row(s) in 0.0280 seconds

hbase(main):006:0> scan 'testtable'
ROW   COLUMN+CELL
0 row(s) in 0.0260 seconds

hbase(main):007:0> put 'testtable', 'row1', 'colfam1:qual1', 'val1',  
  Time.now.to_i - 50000 
0 row(s) in 0.0260 seconds

hbase(main):008:0> scan 'testtable'
ROW   COLUMN+CELL
0 row(s) in 0.0260 seconds

hbase(main):009:0> flush 'testtable' 
0 row(s) in 0.2720 seconds 

hbase(main):010:0> major_compact 'testtable'
0 row(s) in 0.0420 seconds

hbase(main):011:0> put 'testtable', 'row1', 'colfam1:qual1', 'val1',  
  Time.now.to_i - 50000 
0 row(s) in 0.0280 seconds

hbase(main):012:0> scan 'testtable'
ROW   COLUMN+CELL
 row1 column=colfam1:qual1, timestamp=1308900423953, value=val1
1 row(s) in 0.0290 seconds

The example shows that there are sometimes situations where you might see something you do not expect to see. But this behavior is explained by the architecture of HBase, and is deterministic.

Since you can specify your own timestamp values—and therefore create your own versioning scheme—while overriding the server-side timestamp generation based on the synchronized server time, you are free to not use epoch-based versions at all.

For example, you could use the timestamp with a global number generator^[108] that supplies you with ever increasing, sequential numbers starting at 1. Every time you insert a new value you retrieve a new number and use that when calling the put function.

You must do this for every put operation, or the server will insert an epoch-based timestamp instead. There is a flag in the table or column descriptors that indicates your use of custom timestamp values; in other words, your own versioning. If you fail to set the value, it is silently replaced with the server timestamp.

With these warnings out of the way, here are a few use cases that show how a custom versioning scheme can be beneficial in the overall concept of table schema design:

Using the time component of HBase is an interesting way to exploit this extra dimension offered by the architecture. You have less freedom, as it only accepts long values, as opposed to arbitrary binary keys supported by row and column keys. Nevertheless, it could solve your specific use case.