The first set of files are the write-ahead log files handled by the HLog instances, created in a directory called .logs underneath the HBase root directory. The .logs directory contains a subdirectory for each HRegionServer. In each subdirectory, there are several HLog files (because of log rotation). All regions from that region server share the same HLog files.

An interesting observation is that the logfile is reported to have a size of 0. This is fairly typical when the file was created recently, as HDFS is using built-in append support to write to this file, and only complete blocks are made available to readers—including the hadoop dfs -lsr command. Although the data of the put operations is safely persisted, the size of the logfile that is currently being written to is slightly off.

After, for example, waiting for an hour so that the logfile is rolled (see LogRoller Class for all reasons when logfiles are rolled), you will see the existing logfile reported with its proper size, since it is closed now and HDFS can state the “correct” size. The new logfile next to it again starts at zero size:

  249962 /hbase/.logs/foo.internal,60020,1309812147645/ 
foo.internal%2C60020%2C1309812147645.1309812151180
       0 /hbase/.logs/foo.internal,60020,1309812147645/ 
foo.internal%2C60020%2C1309812147645.1309815751223

When a logfile is are no longer needed because all of the contained edits have been persisted into store files, it is decommissioned into the .oldlogs directory under the root HBase directory. This is triggered when the logfile is rolled based on the configured thresholds.

The old logfiles are deleted by the master after 10 minutes (by default), set with the hbase.master.logcleaner.ttl property. The master checks every minute (by default again) for those files. This is configured with the hbase.master.cleaner.interval property.

The hbase.id and hbase.version files contain the unique ID of the cluster, and the file format version:

$ hadoop dfs -cat /hbase/hbase.id
$e627e130-0ae2-448d-8bb5-117a8af06e97
$ hadoop dfs -cat /hbase/hbase.version
7

They are used internally and are otherwise not very interesting. In addition, there are a few more root-level directories that appear over time. The splitlog and .corrupt folders are used by the log split process to store the intermediate split files and the corrupted logs, respectively. For example:

      0 /hbase/.corrupt
      0 /hbase/splitlog/foo.internal,60020,1309851880898_hdfs%3A%2F%2F 
localhost%2Fhbase%2F.logs%2Ffoo.internal%2C60020%2C1309850971208%2F 
foo.internal%252C60020%252C1309850971208.1309851641956/testtable/ 
d9ffc3a5cd016ae58e23d7a6cb937949/recovered.edits/0000000000000002352

There are no corrupt logfiles in this example, but there is one staged split file. The log splitting process is explained in Replay.

Every table in HBase has its own directory, located under the HBase root directory in the filesystem. Each table directory contains a top-level file named .tableinfo, which stores the serialized HTableDescriptor (see Tables for details) for the table. This includes the table and column family schemas, and can be read, for example, by tools to gain insight on what the table looks like. The .tmp directory contains temporary data, and is used, for example, when the .tableinfo file is updated.

Inside each table directory, there is a separate directory for every region comprising the table. The names of these directories are the MD5 hash portion of a region name. For example, the following is taken from the master’s web UI, after clicking on the testtable link in the User Tables section:

testtable,row-500,1309812163930.d9ffc3a5cd016ae58e23d7a6cb937949.

The MD5 hash is d9ffc3a5cd016ae58e23d7a6cb937949 and is generated by encoding everything before the hash in the region name (minus the dividing dot), that is, testtable,row-500,1309812163930. The final dot after the hash is part of the complete region name: it indicates that this is a new style name. In previous versions of HBase, the region names did not include the hash.

.META.,,1.1028785192

The hash guarantees that the directory names are always valid, in terms of filesystem rules: they do not contain any special character, such as the slash (“/”), which is used to divide the path. The overall layout for region files is then:

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

In each column-family directory, you can see the actual data files, explained in HFile Format. Their name is just an arbitrary number, based on the Java built-in random generator. The code is smart enough to check for collisions, that is, where a file with a newly generated number already exists. It loops until it finds an unused one and uses that instead.

The region directory also has a .regioninfo file, which contains the serialized information of the HRegionInfo instance for the given region. Similar to the .tableinfo file, it can be used by external tools to gain insight into the metadata of a region. The hbase hbck tool uses this to generate missing meta table entries, for example.

The optional .tmp directory is created on demand, and is used to hold temporary files—for example, the rewritten files from a compaction. These are usually moved out into the region directory once the process has completed. In rare circumstances, you might find leftover files, which are cleaned out when the region is reopened.

During the replay of the write-ahead log, any edit that has not been committed is written into a separate file per region. These are staged first (see the splitlog directory in Root-level files) and then—assuming the log splitting process has completed successfully—moved into the optional recovered.edits directory atomically. When the region is opened the region server will see the recovery file and replay the entries accordingly.

Once the region needs to split because it has exceeded the maximum configured region size, a matching splits directory is created, which is used to stage the two new daughter regions. If this process is successful—usually this happens in a few seconds or less—they are moved up into the table directory to form the two new regions, each representing one-half of the original region.

In other words, when you see a region directory that has no .tmp directory, no compaction has been performed for it yet. When it has no recovered.edits file, no write-ahead log replay has occurred for it yet.

This concludes the list of what is commonly contained in the various directories inside the HBase root folder. There are more intermediate files, created by the region split process. They are discussed separately in the next section.

When a store file within a region grows larger than the configured hbase.hregion.max.filesize—or what is configured at the column family level using HColumnDescriptor—the region is split in two. This is done initially very quickly because the system simply creates two reference files for the new regions (also called daughters), which each hosting half of the original region (referred to as the parent).

The region server accomplishes this by creating the splits directory in the parent region. Next, it closes the region so that it does not take on anymore requests.

The region server then prepares the new daughter regions (using multiple threads) by setting up the necessary file structures inside the splits directory. This includes the new region directories and the reference files. If this process completes successfully, it moves the two new region directories into the table directory. The .META. table is updated for the parent to state that it is now split, and what the two daughter regions are. This prevents it from being reopened by accident. Here is an example of how this looks in the .META. table:

row: testtable,row-500,1309812163930.d9ffc3a5cd016ae58e23d7a6cb937949. 

  column=info:regioninfo, timestamp=1309872211559, value=REGION => {NAME => 
    'testtable,row-500,1309812163930.d9ffc3a5cd016ae58e23d7a6cb937949. 
     TableName => 'testtable', STARTKEY => 'row-500', ENDKEY => 'row-700', 
     ENCODED => d9ffc3a5cd016ae58e23d7a6cb937949, OFFLINE => true, 
     SPLIT => true,}
                                                                                   
  column=info:splitA, timestamp=1309872211559, value=REGION => {NAME => 
    'testtable,row-500,1309872211320.d5a127167c6e2dc5106f066cc84506f8. 
    TableName => 'testtable', STARTKEY => 'row-500', ENDKEY => 'row-550', 
    ENCODED => d5a127167c6e2dc5106f066cc84506f8,}                                                                                                                       
  column=info:splitB, timestamp=1309872211559, value=REGION => {NAME => 
    'testtable,row-550,1309872211320.de27e14ffc1f3fff65ce424fcf14ae42. 
    TableName => [B@62892cc5', STARTKEY => 'row-550', ENDKEY => 'row-700',  
    ENCODED => de27e14ffc1f3fff65ce424fcf14ae42,}

You can see how the original region was split into two regions, separated at row-550. The SPLIT => true in the info:regioninfo column value also indicates that this region is now split into the regions referred to in info:splitA and info:splitB.

The name of the reference file is another random number, but with the hash of the referenced region as a postfix, for instance:

/hbase/testtable/d5a127167c6e2dc5106f066cc84506f8/colfam1/ 
6630747383202842155.d9ffc3a5cd016ae58e23d7a6cb937949

This reference file represents one-half of the original region with the hash d9ffc3a5cd016ae58e23d7a6cb937949, which is the region shown in the preceding example. The reference files only hold a little information: the key the original region was split at, and whether it is the top or bottom reference. Of note is that these references are then used by the HalfHFileReader class (which was omitted from the earlier overview as it is only used temporarily) to read the original region data files, and either the top or the bottom half of the files.

Both daughter regions are now ready and will be opened in parallel by the same server. This includes updating the .META. table to list both regions as available regions—just like any other. After that, the regions are online and start serving requests.

The opening of the daughters also schedules a compaction for both—which rewrites the store files in the background from the parent region into the two halves, while replacing the reference files. This takes place in the .tmp directory of the daughter regions. Once the files have been generated, they atomically replace the reference.

The parent is eventually cleaned up when there are no more references to it, which means it is removed as the parent from the .META. table, and all of its files on disk are deleted. Finally, the master is informed about the split and can schedule for the new regions to be moved off to other servers for load balancing reasons.

The store files are monitored by a background thread to keep them under control. The flushes of memstores slowly build up an increasing number of on-disk files. If there are enough of them, the compaction process will combine them to a few, larger files. This goes on until the largest of these files exceeds the configured maximum store file size and triggers a region split (see Region splits).

Compactions come in two varieties: minor and major. Minor compactions are responsible for rewriting the last few files into one larger one. The number of files is set with the hbase.hstore.compaction.min property (which was previously called hbase.hstore.compactionThreshold, and although deprecated is still supported). It is set to 3 by default, and needs to be at least 2 or more. A number too large would delay minor compactions, but also would require more resources and take longer once the compactions start.

The maximum number of files to include in a minor compaction is set to 10, and is configured with hbase.hstore.compaction.max. The list is further narrowed down by the hbase.hstore.compaction.min.size (set to the configured memstore flush size for the region), and the hbase.hstore.compaction.max.size (defaults to Long.MAX_VALUE) configuration properties. Any file larger than the maximum compaction size is always excluded. The minimum compaction size works slightly differently: it is a threshold rather than a per-file limit. It includes all files that are under that limit, up to the total number of files per compaction allowed.

Figure 8-4 shows an example set of store files. All files that fit under the minimum compaction threshold are included in the compaction process.

Figure 8-4. A set of store files showing the minimum compaction threshold

The algorithm uses hbase.hstore.compaction.ratio (defaults to 1.2, or 120%) to ensure that it does include enough files in the selection process. The ratio will also select files that are up to that size compared to the sum of the store file sizes of all newer files. The evaluation always checks the files from the oldest to the newest. This ensures that older files are compacted first. The combination of these properties allows you to fine-tune how many files are included in a minor compaction.

In contrast to minor compactions, major compactions compact all files into a single file. Which compaction type is run is automatically determined when the compaction check is executed. The check is triggered either after a memstore has been flushed to disk, after the compact or major_compact shell commands or corresponding API calls have been invoked, or by a background thread. This background thread is called the CompactionChecker and each region server runs a single instance. It runs a check on a regular basis, controlled by hbase.server.thread.wakefrequency (and multiplied by hbase.server.thread.wakefrequency.multiplier, set to 1000, to run it less often than the other thread-based tasks).

If you call the major_compact shell command, or the majorCompact() API call, you force the major compaction to run. Otherwise, the server checks first if the major compaction is due, based on hbase.hregion.majorcompaction (set to 24 hours) from the first time it ran. The hbase.hregion.majorcompaction.jitter (set to 0.2, in other words, 20%) causes this time to be spread out for the stores. Without the jitter, all stores would run a major compaction at the same time, every 24 hours. See Managed Splitting for information on why this is a bad idea and how to manage this better.

If no major compaction is due, a minor compaction is assumed. Based on the aforementioned configuration properties, the server determines if enough files for a minor compaction are available and continues if that is the case.

Minor compactions might be promoted to major compactions when the former would include all store files, and there are less than the configured maximum files per compaction.

Since all edits are written to one HLog-based logfile per region server, you might ask: why is that the case? Why not write all edits for a specific region into its own logfile? Here is the related quote from the Bigtable paper:

HBase followed that principle for pretty much the same reasons: writing too many files at the same time, plus the number of rolled logs that need to be kept, does not scale well.

What is the drawback, though? If you have to split a log because of a server crash, you need to divide it into suitable pieces, as described in the next section. The master cannot redeploy any region from a crashed server until the logs for that very server have been split. This can potentially take a considerable amount of time.

There are two situations in which logfiles have to be replayed: when the cluster starts, or when a server fails. When the master starts—and this includes a backup master taking over duty—it checks if there are any logfiles, in the .logs directory under the HBase root on the filesystem, that have no region server assigned to them. The logs’ names contain not just the server name, but also the start code of the server. This number is reset every time a region server restarts, and the master can use this number to verify whether a log has been abandoned—for example, due to a server crash.

The master is responsible for monitoring the servers using ZooKeeper, and if it detects a server failure, it immediately starts the process of recovering its logfiles, before reassigning the regions to new servers. This happens in the ServerShutdownHandler class.

Before the edits in the log can be replayed, they need to be separated into one logfile per region. This process is called log splitting: the combined log is read and all entries are grouped by the region they belong to. These grouped edits are then stored in a file next to the target region for subsequent recovery.

The actual process of splitting the logs is different in nearly every version of HBase: early versions would read the file in a single thread, directly on the master. This was improved to at least write the grouped edits per region in multiple threads. Version 0.92.0 finally introduces the concept of distributed log splitting, which removes the burden of doing the actual work from the master to all region servers.

Consider a larger cluster with many region servers and many (rather large) logfiles. In the past, the master had to recover each logfile separately, and—so it would not overload in terms of I/O as well as memory usage—it would do this sequentially. This meant that, for any region that had pending edits, it had to be blocked from opening until the log split and recovery had been completed.

The new distributed mode uses ZooKeeper to hand out each abandoned logfile to a region server. They monitor ZooKeeper for available work, and if the master indicates that a log is available for processing, they race to accept the task. The winning region server then proceeds to read and split the logfiles in a single thread (so as not to overload the already busy region server).

The split process writes the edits first into the splitlog staging directory under the HBase root folder. They are placed in the same path that is needed for the target region. For example:

      0 /hbase/.corrupt
      0 /hbase/splitlog/foo.internal,60020,1309851880898_hdfs%3A%2F%2F 
localhost%2Fhbase%2F.logs%2Ffoo.internal%2C60020%2C1309850971208%2F 
foo.internal%252C60020%252C1309850971208.1309851641956/testtable/ 
d9ffc3a5cd016ae58e23d7a6cb937949/recovered.edits/0000000000000002352

The path contains the logfile name itself to distinguish it from other, possibly concurrently executed, log split output. The path also contains the table name, region name (hash), and recovered.edits directory. Lastly, the name of the split file is the sequence ID of the first edit for the particular region.

The .corrupt directory contains any logfile that could not be parsed. This is influenced by the hbase.hlog.split.skip.errors property, which is set to true by default. It means that any edit that could not be read from a file causes the entire log to be moved to the .corrupt folder. If you set the flag to false, an IOExecption is thrown and the entire log splitting process is stopped.

Once a log has been split successfully, the per-region files are moved into the actual region directories. They are now ready to be recovered by the region itself. This is also why the splitting has to stall opening the affected regions, since it first has to provide the pending edits for replay.

When a region is opened, either because the cluster is started or because it has been moved from one region server to another, it first checks for the presence of the recovered.edits directory. If it exists, it opens the contained files and starts reading the edits they contain. The files are sorted by their name, which contains the sequence ID. This allows the region to recover the edits in order.

Any edit that has a sequence ID that is less than or equal to what has been persisted in the on-disk store files is ignored, because it has already been applied. All other edits are applied to the matching memstore of the region to recover the previous state. At the end, a flush of the memstores is forced to write the current state to disk.

The files in the recovered.edits folder are removed once they have been read and their edits persisted to disk. If a file cannot be read, the hbase.skip.errors property defines what happens next: the default value is false and causes the entire region recovery to fail. If this property is set to true, the file is renamed to the original filename plus .<currentTimeMillis>. Either way, you need to carefully check your logfiles to determine why the recovery has had issues and fix the problem to continue.

The client uses an HBase API that sends a Put, Delete, or Increment to a region server. The key/values are transformed into a WALEdit by the region server and the WALEdit is inspected by the replication code that, for each family that is scoped for replication, adds the scope to the edit. The edit is appended to the current WAL and is then applied to its MemStore.

In a separate thread, the edit is read from the log (as part of a batch) and only the KeyValues that are replicable are kept (i.e., they are part of a family that is scoped as GLOBAL in the family’s schema and are noncatalog so it is not .META. or -ROOT-). When the buffer is filled, or the reader hits the end of the file, the buffer is sent to a random region server on the slave cluster.

Synchronously, the region server that receives the edits reads them sequentially and separates each of them into buffers, one per table. Once all edits are read, each buffer is flushed using the normal HBase client (HTables managed by an HTablePool). This is done in order to leverage parallel insertion (MultiPut).

Back in the master cluster’s region server, the offset for the current WAL that is being replicated is registered in ZooKeeper.

The edit is inserted in the same way. In a separate thread, the region server reads, filters, and buffers the log edits the same way as is done during normal processing. The slave region server that is contacted does not answer to the RPC, so the master region server will sleep and retry up to a configured number of times. If the slave region server still is not available, the master cluster region server will select a new subset of the region server to replicate to and will try to send the buffer of edits again.

In the meantime, the WALs will be rolled and stored in a queue in ZooKeeper. Logs that are archived by their region server (archiving is basically moving a log from the region server’s logs directory to a central logs archive directory) will update their paths in the in-memory queue of the replicating thread.

When the slave cluster is finally available, the buffer will be applied the same way as during normal processing. The master cluster region server will then replicate the backlog of logs.

When a master cluster region server initiates a replication source to a slave cluster, it first connects to the slave’s ZooKeeper ensemble using the provided cluster key (that key is composed of the value of hbase.zookeeper.quorum, zookeeper.znode.parent, and hbase.zookeeper.property.clientPort). It then scans the /hbase/rs directory to discover all the available sinks (region servers that are accepting incoming streams of edits to replicate) and will randomly choose a subset of them using a configured ratio (which has a default value of 10%). For example, if a slave cluster has 150 machines, 15 will be chosen as potential recipients for edits that this master cluster region server will be sending. Since this is done by all master cluster region servers, the probability that all slave region servers are used is very high, and this method works for clusters of any size. For example, a master cluster of 10 machines replicating to a slave cluster of five machines with a ratio of 10% means that the master cluster region servers will choose one machine each at random; thus the chance of overlapping and full usage of the slave cluster is higher.

Every master cluster region server has its own znode in the replication znodes hierarchy. The parent znode contains one znode per peer cluster (if there are five slave clusters, five znodes are created), and each of these contains a queue of HLogs to process. Each of these queues will track the HLogs created by that region server, but they can differ in size. For example, if one slave cluster becomes unavailable for some time, the HLogs should not be deleted, and thus they need to stay in the queue (while the others are processed). See Region server failover for an example.

When a source is instantiated, it contains the current HLog that the region server is writing to. During log rolling, the new file is added to the queue of each slave cluster’s znode just before it is made available. This ensures that all the sources are aware that a new log exists before HLog is able to append edits into it, but this operation is now more expensive. The queue items are discarded when the replication thread cannot read more entries from a file (because it reached the end of the last block) and that there are other files in the queue. This means that if a source is up-to-date and replicates from the log that the region server writes to, reading up to the “end” of the current file will not delete the item in the queue.

When a log is archived (because it is not used anymore or because there are too many of them per hbase.regionserver.maxlogs, typically because the insertion rate is faster than the region flushing rate), it will notify the source threads that the path for that log changed. If a particular source was already done with it, it will just ignore the message. If it is in the queue, the path will be updated in memory. If the log is currently being replicated, the change will be done atomically so that the reader does not try to open the file when it is already moved. Also, moving a file is a NameNode operation; so, if the reader is currently reading the log, it will not generate any exceptions.

By default, a source will try to read from a logfile and ship log entries as quickly as possible to a sink. This is first limited by the filtering of log entries; only KeyValues that are scoped GLOBAL and that do not belong to catalog tables will be retained. A second limit is imposed on the total size of the list of edits to replicate per slave, which by default is 64 MB. This means that a master cluster region server with three slaves will use, at most, 192 MB to store data to replicate. This does not take into account the data that was filtered but was not garbage-collected.

Once the maximum number of edits has been buffered or the reader has hit the end of the logfile, the source thread will stop reading and will randomly choose a sink to replicate to (from the list that was generated by keeping only a subset of slave region servers). It will directly issue an RPC to the chosen machine and will wait for the method to return. If it is successful, the source will determine if the current file is emptied or if it should continue to read from it. If the former, it will delete the znode in the queue. If the latter, it will register the new offset in the log’s znode. If the RPC threw an exception, the source will retry 10 times until trying to find a different sink.

If replication is not enabled, the master’s log cleaning thread will delete old logs using a configured TTL. This does not work well with replication since archived logs that are past their TTL may still be in a queue. Thus, the default behavior is augmented so that if a log is past its TTL, the cleaning thread will look up every queue until it finds the log (while caching the ones it finds). If it is not found, the log will be deleted. The next time it has to look for a log, it will first use its cache.

As long as region servers do not fail, keeping track of the logs in ZooKeeper does not add any value. Unfortunately, they do fail, so since ZooKeeper is highly available, we can count on it and its semantics to help us manage the transfer of the queues.

All the master cluster region servers keep a watcher on one another to be notified when one dies (just like the master does). When this happens, they all race to create a znode called lock inside the dead region server’s znode that contains its queues. The one that creates it successfully will proceed by transferring all the queues to its own znode (one by one, since ZooKeeper does not support the rename operation) and will delete all the old ones when it is done. The recovered queues’ znodes will be named with the ID of the slave cluster appended with the name of the dead server.

Once that is done, the master cluster region server will create one new source thread per copied queue, and each of them will follow the read/filter/ship pattern. The main difference is that those queues will never have new data since they do not belong to their new region server, which means that when the reader hits the end of the last log, the queue’s znode will be deleted and the master cluster region server will close that replication source.

For example, consider a master cluster with three region servers that is replicating to a single slave with an ID of 2. The following hierarchy represents what the znodes’ layout could be at some point in time. We can see that the region servers’ znodes all contain a peers znode that contains a single queue. The znode names in the queues represent the actual filenames on HDFS in the form address,port.timestamp.

/hbase/replication/rs/
                      1.1.1.1,60020,123456780/
                        peers/
                              2/
                                1.1.1.1,60020.1234  (Contains a position)
                                1.1.1.1,60020.1265
                      1.1.1.2,60020,123456790/
                        peers/
                              2/
                                1.1.1.2,60020.1214  (Contains a position)
                                1.1.1.2,60020.1248
                                1.1.1.2,60020.1312
                      1.1.1.3,60020,    123456630/
                        peers/
                              2/
                                1.1.1.3,60020.1280  (Contains a position)

Now let’s say that 1.1.1.2 loses its ZooKeeper session. The survivors will race to create a lock, and for some reason 1.1.1.3 wins. It will then start transferring all the queues to its local peers znode by appending the name of the dead server. Right before 1.1.1.3 is able to clean up the old znodes, the layout will look like the following:

/hbase/replication/rs/
                      1.1.1.1,60020,123456780/
                        peers/
                              2/
                                1.1.1.1,60020.1234  (Contains a position)
                                1.1.1.1,60020.1265
                      1.1.1.2,60020,123456790/
                        lock
                        peers/
                              2/
                                1.1.1.2,60020.1214  (Contains a position)
                                1.1.1.2,60020.1248
                                1.1.1.2,60020.1312
                      1.1.1.3,60020,123456630/
                        peers/
                              2/
                                1.1.1.3,60020.1280  (Contains a position)

                              2-1.1.1.2,60020,123456790/
                                1.1.1.2,60020.1214  (Contains a position)
                                1.1.1.2,60020.1248
                                1.1.1.2,60020.1312

Sometime later, but before 1.1.1.3 is able to finish replicating the last HLog from 1.1.1.2, let’s say that it dies too (also, some new logs were created in the normal queues). The last region server will then try to lock 1.1.1.3’s znode and will begin transferring all the queues. The new layout will be:

/hbase/replication/rs/
                      1.1.1.1,60020,123456780/
                        peers/
                              2/
                                1.1.1.1,60020.1378  (Contains a position)

                              2-1.1.1.3,60020,123456630/
                                1.1.1.3,60020.1325  (Contains a position)
                                1.1.1.3,60020.1401

                              2-1.1.1.2,60020,123456790-1.1.1.3,60020,123456630/
                                1.1.1.2,60020.1312  (Contains a position)
                      1.1.1.3,60020,123456630/
                        lock
                        peers/
                              2/
                                1.1.1.3,60020.1325  (Contains a position)
                                1.1.1.3,60020.1401

                              2-1.1.1.2,60020,123456790/
                                1.1.1.2,60020.1312  (Contains a position)

Replication is still considered to be an experimental feature. Carefully evaluate whether it works for your use case before you consider using it.