HBase, HDFS and durable sync

HBase and HDFS go hand in hand to provide HBase's durability and consistency guarantees.

One way of looking at this setup is that HDFS handles the distribution and storage of your data whereas HBase handles the distribution of CPU cycles and provides a consistent view of that data.

As described in many other places, HBase

1. Appends all changes to a WAL
2. Batches/sorts changes in memory
3. Flushes memory to immutable, sorted data files in HDFS
4. Combines smaller data files to fewer, larger ones during compactions.

The part that is typical less clearly understood and documented are the exact durability guarantees provided by HDFS (and hence HBase).

HDFS sync has a colorful history, with needed support for HBase only available in an unreleased "append"-branch of HDFS for a long time. (Note that the append and sync features are independent and against common believe HBase only relies on the sync feature). See also this Cloudera blog post.

In order to understand what HDFS provides let's take a look at how a DFSClient (client) interacts with a Datanode (DN).

In a nutshell a DN just waits for commands. One of these commands is WRITE_BLOCK. When the DN receives a WRITE_BLOCK command it instantiates a BlockReceiver thread.

The BlockReceiver then simply waits for packets on an InputStream and flushes the data to OS buffers. An open block is maintained at the DN as an open file. When a block is filled, the block and hence its associate files is closed and the BlockReceiver ends. For all practical purposes the DN forgets that the block existed.

Replication to replica DNs is done via pipelining. The first DN forwards each packet to the next DN in the chain before the data is flushed locally, and waits for the downstreadm DN to respond. The default length of the replication chain is 3.

The other side of the equation is the DFSClient. The DFSClient batches changes until a packet is filled.
Since HADOOP-6313 a Syncable supports hflush and hsync.

• hflush flushes all outstanding data (i.e. the current unfinished packet) from the client into the OS buffers on all DN replicas.
• hsync flushes the data to the DNs like hflush and should also force the data to disk via fsync (or equivalent). But currently for HDFS hsync is implemented as hflush!

The gist is that both closing of a block and issuing hsync/hflush on the client currently only guarantees that data was flushed from the client to the replica DNs and to OS buffers on each DN; not that the data actually reached a physical disk.

For HBase and similar applications with durability guarantees this can be insufficient. If three or more DN machines crash at the same time (assume three replicas), for example caused by a multi rack or data center power outage, data might be lost.
Further, since HBase constantly compacts older, smaller HFiles into newer, larger ones, this potential data loss is not limited to new data.
(But note that, like most database setups, HBase should be deployed with redundant power supply anyway, so this is not necessarily an issue).

Due to the inner working of the DN it is difficult to implement 100% Posix fsync semantics. Imagine a client, which writes many blocks worth of data and then issues an hsync. In order to sync data correctly to disk either the client or all involved DNs would need to keep track of all blocks (full or partial) written to so far that have not yet been sync'ed.

This would be a significant change to how either the client or the DN work and lead to more complicated code. It would also require to keep the block files open in order to retain the file descriptors so that an fsync could be potentially issued in the future. Since the client might in fact never issue a sync request the number of open files to retain is unbounded.

The other option (similar to Posix' O_SYNC) is to have the DNs call fsync upon receipt of every single packet. leading to many unnecessary fsyncs.

In HDFS-744 I propose a hybrid solution. A data stream can be created with a SYNC_BLOCK flag. This flag causes the DFSClient set a "sync" flag on the last packet of a block. I.e. the block file is fsync'ed upon close.

This flag is also set when the client issues hsync. If the client has outstanding data the current packet is tagged with the "sync" flag and sent immediately, otherwise an empty packet with this flags is sent.

When a DN receives such a packet, it will immediately flush the currently open file (representing the current block - full on close or partial on hsync - being written) to disk.

In summary: With this compromise a client can guarantee - byte-by-byte if needed - which portion of an open file is guaranteed on a durable medium while avoiding either syncing every packet to disk or keeping track of past unsync'ed block.
For HBase this would conveniently deal with compactions as blocks are sync'ed upon close and also with WAL edits as it correctly allows sync'ing the current block.

The downside is that upon close each block needs to be sync'ed to disk, even though the client might never issue a sync request for this stream; this leads to potentially unneeded fsyncs.

HBASE-5954 proposes matching changes to HBase to make use of this new HDFS feature. This issue introduces a WAL sync config option and an HFile sync option.
The former causes HBase to issue an hsync when a batch of WAL entries is written. The latter makes sure HFiles (generated from memstore flushes or compactions) are guaranteed to be on a durable medium when the stream is closed.

There are also a few simple performance tests listed in that issue.

Future optimization is possible:

• Only one of the replica DNs could issue the sync
• Only one DN in each Rack could issue the sync
• The sync could be done in parallel and the response from the DN need not wait for for it to finish (in this case the client has no guarantee that the sync actually finished when hsync returns, only that the DN promised to do it)
• For HBase, both options could be made configurable per column family.

HBaseCon

HBaseCon on May 22nd was a blast.

I hosted the development track and presented on HBase Internals (which is essentially a condensed version of this blog). And that all despite my being mortally afraid of public speaking.

The slides of the Internals talk are now available.
Slideshare does not deal well with animation, unfortunately.

I'll add a link to the video once it becomes available.

HBase 0.94

In all the hectic leading up to HBaseCon, I forgot to blog about HBase 0.94.

I released HBase 0.94.0 on May 15, 2012. This was the 4th release candidate. 

0.94 is mostly a performance release with a lot of contributions from Facebook.
As mentioned in the release announcement these are some notable improvements:

• HBASE-5010 - Filter HFiles based on TTL
• HBASE-4465 - Lazy-seek optimization for StoreFile scanners
• HBASE-4469 - Avoid top row seek by looking up ROWCOL bloomfilter
• HBASE-4532 - Avoid top row seek by dedicated bloom filter for delete family bloom filter
• HBASE-5074 - support checksums in HBase block cache

among others. Many thanks to Mikhail Bautin and others at Facebook for all these improvements.

Of course there are some new features as well.

• Jonathan Hsieh's HBASE-5128 - [uber hbck] Online automated repair of table integrity and region consistency problems
• Jon Gray's HBASE-4460 - Support running an embedded ThriftServer within a RegionServer

And then of course my pets (I blogged about these before):

• HBASE-3584 - Allow atomic put/delete in one call
• HBASE-4102 - atomicAppend: A put that appends to the latest version of a cell
• HBASE-4536 - Allow CF to retain deleted rows
• HBASE-5229 - Provide basic building blocks for "multi-row" local transactions.

0.94 is binary compatible with 0.92. Server and client can be freely mixed between versions, it is even possible to mix servers of different version within the same cluster (for rolling upgrades).

I expect 0.94.x to be used for a while until the "singularity" (0.96) stabilizes into HBase 1.0.

Update Wednesday, May 30th, 2012:
HBase can be downloaded from here: http://www.apache.org/dyn/closer.cgi/hbase/
The list of features and bug fixes is here.

(timestamp-) Consistent backups in HBase

The topic consistent backups in HBase comes up every now and then.
In this article I will outline a scheme that does provide timestamp-consistent backups.

Consistent backups are possible in HBase. With "consistent" I mean "consistent as of a specific timestamp" (this is limited by the HBase timestamp granularity.)

Over the past few months I have contributed various changes to HBase that now hopefully lead full circle to a more coherent story for "system of record" type data retention.

The basic setup is simple. You'll need

• HBASE-4071 (0.92 - for MIN_VERSIONS)
• HBASE-4536 (0.94 - keeping deleted cells and raw scans)
• HBASE-4682 (0.94 - deleted cells in exports)

(HBase 0.94 will have all the relevant patches once it is releases - which should be any time now)

HBase's built-in Export tool can then be used to generate consistent snapshots.

Normally the data collected by an Export job is "smeared" over the time interval it takes to execute the job; an Export-Scan sees a row (cells of a row to be precise) as of the time when it happens to get to it, and rows can change while the Export is running.

That is problematic, because it is not possible to recreate a consistent view of the database from export generated this way.

The trick then is to keep all data in HBase long enough for the backup job to finish, and to only collect information before the start time of the export job.

Say the backup takes no more than T to complete (yes, it is hard to know ahead of time how long an Export is going to run). In that case the table's column families can be setup as follows:

• set TTL to T + some headroom (so maybe 2T to be safe)
• set VERSIONS to a very large number, (max int = 2147483647 for example, i.e. nothing is evicted from HBase due to VERSIONS)
• set MIN_VERSIONS to how many versions you want to keep around, otherwise all versions could be removed if their TTL expired
• set KEEP_DELETED_CELLS to true (this makes sure that deleted cell and delete markers are kept until they expire by the TTL)

In the shell (setting TTL to two hours):

create <table>, {NAME=><columnfamily>, VERSIONS=>2147483647, TTL=>7200, MIN_VERSIONS=>1, KEEP_DELETED_CELLS=>true}

Export can now we used as follows:

• set versions to 2147483647 (i.e. all versions)
• set startTime to -2147483648 (min int, i.e. everything is guaranteed to be included)
• set endTime to the start time of the Export (i.e. the current time when the export starts. This is the crucial part)
• enable support for deleted cell

hbase org.apache.hadoop.hbase.mapreduce.Export
-D hbase.mapreduce.include.deleted.rows=true
<tablename> <outputdir>
2147483647 -2147483648 <now in ms>

As long as the Export finishes within 2T, a consistent snapshot as of the time the Export was started is created. Otherwise some data might be missing, as it could have been compacted away before the Export had a chance to see it.

Since the backups also copied deleted rows and delete markers, a backup restored to an HBase instance can be queried using a time range (see Scan) to retrieve the state of the data at any arbitrary time.

Export is current limited to a single table, but given enough storage in your live cluster this can be extended to multiple table Exports, simply by setting the endTime of all Exports jobs to the start time of the first job.


This same trick can also be used for incremental backups. In that case the TTL has to be large enough to cover the interval between incremental backups.
If, for example, the incremental backups frequency is daily, the TTL above can be set to 2 days (TTL=>172800). Then use Export again:

hbase org.apache.hadoop.hbase.mapreduce.Export
-D hbase.mapreduce.include.deleted.rows=true
<tablename> <outputdir>
2147483647 <time of last backup in ms> <now in ms>

The longer TTL guarantees that there will be no gaps that are not covered by the incremental backups.

An example:

1. A Put (p1) happens at T1
2. Full backup starts at T2, time interval [0, T2)
3. Another Put (p2) at T3
4. full backup jobs finishes
5. A Delete happens at T4 
6. Incremental backup starts at T5, time interval [T2, T5)
7. Yet another Put (p3)
8. Incremental backup finishes

Note that in this scenario is does not matter when the backup jobs finish.

The full backup contains only p1. The incremental backup contains p2 and the Delete. p3 is not included in any backup, yet.
The state at T2 (p1) and T5 (p1, p2, delete) can be directly restored. Using time range Scans or Gets the state as of T4 and T3 can also be retrieved, once both backups have been restored into the same HBase instance (you need HBASE-4536 for this to work correctly with Deletes).

Finally, if keeping enough data to cover the time between two incremental backups in the live HBase cluster is problematic for your organization, it is also possible to archive HBase's Write Ahead Logs (WAL) and then replay with the built-in WALPlayer (HBASE-5604), but that is for another post.

ACID in HBase

By Lars Hofhansl

As we know, ACID stands for Atomicity, Consistency, Isolation, and Durability.

HBase supports ACID in limited ways, namely Puts to the same row provide all ACID guarantees. (HBASE-3584 adds multi op transactions and HBASE-5229 adds multi row transactions, but the principle remains the same)

So how does ACID work in HBase?

HBase employs a kind of MVCC. And HBase has no mixed read/write transactions.

The nomenclature in HBase is bit strange for historical reasons. In a nutshell each RegionServer maintains what I will call "strictly monotonically increasing transaction numbers".

When a write transaction (a set of puts or deletes) starts it retrieves the next highest transaction number. In HBase this is called a WriteNumber.
When a read transaction (a Scan or Get) starts it retrieves the transaction number of the last committed transaction. HBase calls this the ReadPoint.

Each created KeyValue is tagged with its transaction's WriteNumber (this tag, for historical reasons, is called the memstore timestamp in HBase. Note that this is separate from the application-visible timestamp.)

The highlevel flow of a write transaction in HBase looks like this:

1. lock the row(s), to guard against concurrent writes to the same row(s)
2. retrieve the current writenumber
3. apply changes to the WAL (Write Ahead Log)
4. apply the changes to the Memstore (using the acquired writenumber to tag the KeyValues)
5. commit the transaction, i.e. attempt to roll the Readpoint forward to the acquired Writenumber.
6. unlock the row(s)

The highlevel flow of a read transaction looks like this:

1. open the scanner
2. get the current readpoint
3. filter all scanned KeyValues with memstore timestamp > the readpoint
4. close the scanner (this is initiated by the client)

In reality it is a bit more complicated, but this is enough to illustrate the point. Note that a reader acquires no locks at all, but we still get all of ACID.

It is important to realize that this only works if transactions are committed strictly serially; otherwise an earlier uncommitted transaction could become visible when one that started later commits first. In HBase transaction are typically short, so this is not a problem.

HBase does exactly that: All transactions are committed serially.

Committing a transaction in HBase means settting the current ReadPoint to the transaction's WriteNumber, and hence make its changes visible to all new Scans.
HBase keeps a list of all unfinished transactions. A transaction's commit is delayed until all prior transactions committed. Note that HBase can still make all changes immediately and concurrently, only the commits are serial.

Since HBase does not guarantee any consistency between regions (and each region is hosted at exactly one RegionServer) all MVCC data structures only need to be kept in memory on every region server.

The next interesting question is what happens during compactions.

In HBase compactions are used to join multiple small store files (create by flushes of the MemStore to disk) into a larger ones and also to remove "garbage" in the process.
Garbage here are KeyValues that either expired due to a column family's TTL or VERSION settings or were marked for deletion. See here and here for more details.

Now imagine a compaction happening while a scanner is still scanning through the KeyValues. It would now be possible see a partial row (see here for how HBase defines a "row") - a row comprised of versions of KeyValues that do not reflect the outcome of any serializable transaction schedule.

The solution in HBase is to keep track of the earliest readpoint used by any open scanner and never collect any KeyValues with a memstore timestamp larger than that readpoint. That logic was - among other enhancements - added with HBASE-2856, which allowed HBase to support ACID guarantees even with concurrent flushes.
HBASE-5569 finally enables the same logic for the delete markers (and hence deleted KeyValues).

Lastly, note that a KeyValue's memstore timestamp can be cleared (set to 0) when it is older than the oldest scanner. I.e. it is known to be visible to every scanner, since all earlier scanner are finished.

Update Thursday, March 22: 
A couple of extra points:

• The readpoint is rolled forward even if the transaction failed in order to not stall later transactions that waiting to be committed (since this is all in the same process, that just mean the roll forward happens in a Java finally block).
• When updates are written to the WAL a single record is created for the all changes. There is no separate commit record.
• When a RegionServer crashes, all in flight transaction are eventually replayed on another RegionServer if the WAL record was written completely or discarded otherwise.

(limited) cross row transactions in HBase

Atomic operations in HBase are currently limited to single row operations. This is achieved by co-locating all cells with the same row-key (see introduction-to-hbase) in the same region.

The key observation is that all that is required for atomic operations are co-located cells.

HBASE-5304 allows defining more flexible RegionSplitPolicies (note Javadoc is not yet updated, so this link leads to nowhere right now).
HBASE-5368 provides an example (KeyPrefixRegionSplitPolicies) that co-locates cells with the same row-key-prefix in the same region.
By choosing the prefix smartly, useful ranges of rows can be co-located together. Obviously one needs to be careful, as regions cannot grow without bounds.

HBASE-5229 finally allows cross row atomic operations over multiple rows as long as they are co-located in the same region.
This is a developer-only feature and only accessible through coprocessor endpoints. However, HBASE-5229 also includes an endpoint that can be used directly.

An example can be setup as follows:
1. add this to hbase-site.xml:
  <property>
    <name>hbase.coprocessor.user.region.classes</name>
    <value>org.apache.hadoop.hbase.coprocessor.MultiRowMutationEndpoint</value>
  </property>

This loads the necessary coprocessor endpoint into all regions of all user tables.

2. Create a table that uses KeyPrefixRegionSplitPolicy:
HTableDescriptor myHtd = new HTableDescriptor("myTable"); myHtd.setValue(HTableDescriptor.SPLIT_POLICY,
               KeyPrefixRegionSplitPolicy.class.getName()); 
// set the prefix to 3 in this example
myHtd.setValue(KeyPrefixRegionSplitPolicy.PREFIX_LENGTH_KEY, 
               String.valueOf(3));

HColumnDescriptor hcd = new HColumnDescriptor(...);
myHtd.addFamily(hcd);

HBaseAdmin admin = ...;

admin.createTable(myHtd);

Regions of "myTable" are now split according to KeyPrefixRegionSplitPolicy with a prefix of 3.


3. Execute an atomic multirow transaction:

List<Mutation> ms = new ArrayList<Mutation>();
Put p = new Put(Bytes.toBytes("xxxabc"));

...
ms.add(p);
Put p1 = new Put(Bytes.toBytes("xxx123"));

...

ms.add(p1);
Delete d = new Delete(Bytes.toBytes("xxxzzz"));

...

ms.add(d);
// get a proxy for MultiRowMutationEndpoint
// Note that the passed row is used to locate the 
// region. Any of the other row keys could have
// been used as well, as long as they identify 
// the same region.
MultiRowMutationProtocol mr = t.coprocessorProxy(
   MultiRowMutationProtocol.class,
   Bytes.toBytes("xxxabc"));
// perform the atomic operation
mr.mutateRows(ms);
 
Update, Saturday, February 18, 2012:
In fact using just "xxx" in the call to get the coprocessorProxy would be correct as well, since it identifies he same region. "xxx" could be considered the "partition" key.

That's it... The two puts and the delete are executed atomically, even though these are different rows.
KeyPrefixRegionSplitPolicy ensures that the three rows are co-located, because they share the same 3-byte prefix.
If any of the involved row-keys is located in a different region, the operation will fail, the client will not try again.
Again, care must be taken to pick a useful prefix length for co-location here, or regions will grow out of bounds, or at least be very skewed in size. 

This feature is useful in certain multi-tenant setups, or for time-based data sets (for example the split boundaries could be set to whole hours, or days).

Filters in HBase (or intra row scanning part II)

Filters in HBase are a somewhat obscure and under-documented feature. (Even us committers are often not aware of their usefulness - see HBASE-5229, and HBASE-4256... Or maybe it's just me...).

Intras row scanning can be done using ColumnRangeFilter. Other filters such as ColumnPrefixFilter or MultipleColumnPrefixFilter might also be handy for this. All three filters have in common that they can provide scanners (see scanning in hbase) with what I will call "seek hints". These hints allow a scanner to seek to the next column, the next row, or an arbitrary next cell determined by the filter. This is far more efficient than having a dumb filter that is passed each cell and determines whether the cell is included in the result or not.

Many other filters also provide these "seek hints". The exception here are filters that filter on column values, as there is no inherent ordering between column values; these filters need to look at the value for each column.

For example check out this code in MultipleColumnPrefixFilter (ASF 2.0 license):
    TreeSet<byte []> lesserOrEqualPrefixes =
      (TreeSet<byte []>) sortedPrefixes.headSet(qualifier, true);
    if (lesserOrEqualPrefixes.size() != 0) {
      byte [] largestPrefixSmallerThanQualifier = lesserOrEqualPrefixes.last();
      if (Bytes.startsWith(qualifier, largestPrefixSmallerThanQualifier)) {
        return ReturnCode.INCLUDE;
      }
      if (lesserOrEqualPrefixes.size() == sortedPrefixes.size()) {
        return ReturnCode.NEXT_ROW;
      } else {
        hint = sortedPrefixes.higher(largestPrefixSmallerThanQualifier);
        return ReturnCode.SEEK_NEXT_USING_HINT;
      }
    } else {
      hint = sortedPrefixes.first();
      return ReturnCode.SEEK_NEXT_USING_HINT;
    }
(the <hint> is used later to skip ahead to that column prefix)

See how this code snippet allows the filter to

1. seek to the next row if all prefixes are know to be less or equal the current qualifier (and the largest didn't match the passed column qualifier). Note that a single seek to the next row can potentially skip millions of columns with a single seek operation.
2. seek to the next larger prefix if there are more prefixes, but the current does not match the qualifier.
3. seek to the first prefix (the smallest) if none the prefixes are less or equal to the current qualifier.

If you didn't feel like looking at the code, you can take away from this that these filters can be safely and efficiently used in very wide rows. If the filter instead would indicate only INCLUDE or SKIP and be forced to visit/examine every version of every column of every row, it would be inefficient to use for wide rows with hundreds of thousands or millions of columns.

I'm in the process of adding more information for these Filter to the HBase Book Reference Guide.