Misconfiguration

In previous HBase versions, 1 GB was the default recommendation for region size. HBase can now easily handle regions ranging in size from 10 to 40 GB, thanks to optimizations in HBase and ever-increasing default memory size in hardware. For more details on how to run HBase with more memory, refer to Chapter 17. When an older HBase cluster is migrated into a recent version (0.98 or greater), a common mistake is to retain previous configuration settings. As a result, HBase tries to keep regions under 1 GB when it will be better letting them grow up to 10 GB or more. Also, even if the cluster has not been migrated from an older version, the maximum region size configuration might have been mistakenly modified to a smaller value, resulting in too many regions.

Misoperation

There are also multiple manual operations that might cause HBase to have too many regions to serve.

Before 0.98

Before version 0.98, there are two main options to merge regions together: copying the table to a new presplit table or merging regions. Before HBase 0.98, regions could be merged offline only.

The first option is to create a new table with the same definitions as the existing table but with a different table name. Copy the data from the original table into this table. To complete the process, drop the original table and then rename the new table to the original table name. In this method, we will use the HBase snapshot feature so HBase 0.94.6 or above is required. Also, you will need to run the MapReduce CopyTable job on top of the HBase cluster, so you will need to make sure that the MapReduce framework is available and running. Last, this method will require a temporary suspension of data ingestion. This interruption will be short but is the important part of the process.

The new table should be presplit with the expected newly merged region boundaries. Figure 14-2 illustrates this. Once the presplit step is completed, use the CopyTable command to copy data from the original table to the new table. The CopyTable tool accepts a date range as an argument and will then copy data from time t0 to t1 directly into the target table, into the correct regions. This will allow you to transfer the oldest data first and rerun the tool again to transfer newly created data. CopyTable also allows you to rename or drop unnecessary column families, which can be very useful in case of table redesign.

Figure 14-1. Risk when performing updates with timestamp of cells in the past

Figure 14-2 is an example to better understand how this can be done. Consider that you built an application that stores transactions into a table using the transaction ID as the row key. Your initial assumption was that the transaction ID would be a unique randomly generated readable string. Based on this assumption, the initial table was split into 26 regions from A to Z. However, after running the application for a few weeks, you notice that the transaction ID is a hexadecimal ID. Therefore, instead of going from A to Z, it goes from 0 to F. With the new distribution, regions greater than F will never be used, while the first region will get way more data than the others (puts from 0 to A) and will split into many new regions as it grows. In this case, we’re only moving 26 region to 8 regions; but if you scale this example, you can easily see how hundreds of regions may have been initially created to handle more load and how the same issue of data balancing can occur.

Figure 14-2. Regions redesign

In order to fix this, we will create a new table where regions are already presplit on hexadecimal boundaries. Using the CopyTable command, we will copy data from the existing table to the new table, from t0 to the current day:

hbase org.apache.hadoop.hbase.mapreduce.CopyTable --starttime=0 
--endtime=1411264757000 --new.name=new_table previous_table

The goal of the starttime and endtime parameters is to make sure we copy a specific and controlled section of the source table. Because this operation might take some time, before going to the next steps, and to reduce the input table downtime, we want to copy as much data as possible; but we also want to make sure we will be able to restart the copy from a specific point in time.

Depending on the size of your dataset, this copy might take several hours to a few days. While the copy is running, normal operations can continue on the cluster and on the initial table.

Once the first CopyTable operation is done, the new table contains almost all of the existing data except what has been added between when your previous operation was started (t=141126475700, in this example) and now. Depending on the time it takes to run the first copy, this can represent a considerable amount of data. Because of that, and to make sure the next steps are going to be as short as possible, we will rerun the CopyTable command again. Because the endtime parameter that we used in the previous command is exclusive, we will use it for our current starttime:

hbase org.apache.hadoop.hbase.mapreduce.CopyTable --starttime=1411264757000 
--endtime=1411430414000 --new.name=new_table previous_table

This second operation should be faster, as most of the data was copied in the initial operation. When the two CopyTable operations are done and before doing the last run, to not lose any data, you will have to stop the ingestion to your existing table while you are performing the following operations. In production environments, it is sometimes very sensitive to stop data ingestion. If that is the case, you can think about mitigation solutions. One of them consists of pushing the data into the newly created table. Another is to design the ingestion such that it can accept those pauses. Applications such as Kafka and Flume can be used to queue the data being ingested to HBase and can be paused without losing any information. Typically, the remaining steps might be accomplished in less than one hour. It is hard to quantify how long it will take to bring in the delta. It can be computed from the previous runs, but it can vary widely.

Finally, you will need to do one last run of the CopyTable tool to copy the remaining data. The time between the end of the last run and when you disabled the data ingestion is only a couple minutes, so it should run even faster. This time, because we want to copy all the remaining data, we will again use the last endtime value as the starttime parameter, but we will not specify any endtime:

hbase org.apache.hadoop.hbase.mapreduce.CopyTable --starttime=1411264757000 
--new.name=new_table previous_table

This should run for a few seconds up to a few minutes. At the end of this operation, we will have a duplicate of the original table in our new table, with the new regions boundaries. The last step is to run some snapshot commands to rename the newly created table:

disable 'events_table'
drop 'events_table'
disable 'new_events_table'
snapshot 'new_events_table', 'snap1'
clone_snapshot 'snap1', 'events_table'
delete_snapshot 'snap1'
drop 'new_events_table'

At the end of all those operations, you will have a table containing the exact same data but with a reduced number of regions.

bin/hbase org.apache.hadoop.hbase.util.Merge testtable 
                  "testtable,,1411948096799.77873c05283fe40822ba69a30b601959."
          "testtable,11111111,1411948096800.e7e93a3545d36546ab8323622e56fdc4."

  public int mergeRegion (Configuration config, String tableName,
                          String region1, String region2) {
    String[] args = {tableName, region1, region2};
    int status = -1;
    try {
      status = ToolRunner.run(config, 
      new org.apache.hadoop.hbase.util.Merge(), args);
    } catch (Exception e) {
      e.printStackTrace();
    }
    return status;
  }

Starting with 0.98

HBASE-7403 introduces online merge into HBase starting at version 0.98. Online merge allows you to concatenate two regions into a single one without having to shut down the HBase cluster or disable the target table. This is a big improvement in the merge process. Assuming a high enough version of HBase, this is currently the preferred way to solve the issue of too many regions.

merge_region 'ENCODED_REGIONNAME', 'ENCODED_REGIONNAME'

hbase(main):004:0> merge_region 'cec1ed0e20002c924069f9657925341e',
                                '1d61869389ae461a7971a71208a8dbe5'
0 row(s) in 0.0140 seconds

    Configuration conf = HBaseConfiguration.create();
    Connection connection = ConnectionFactory.createConnection(conf);
    HBaseAdmin admin = (HBaseAdmin)connection.getAdmin();
    List<HRegionInfo> regions = admin.getTableRegions(TableName.valueOf("t1")); 
    LOG.info("testtable contains " + regions.size() + " regions.");
    for (int index = 0; index < regions.size() / 2; index++) {
      HRegionInfo region1 = regions.get(index*2);
      HRegionInfo region2 = regions.get(index*2+1);
      LOG.info("Merging regions " + region1 + " and " + region2);
      admin.mergeRegions(region1.getEncodedNameAsBytes(),
                         region2.getEncodedNameAsBytes(), false); 
    }
    admin.close();

2014-09-24 16:38:59,686 INFO  [main] ch18.Merge: testtable contains 9 regions.
2014-09-24 16:38:59,686 INFO  [main] ch18.Merge: Merging regions ... and ...
2014-09-24 16:38:59,710 INFO  [main] ch18.Merge: Merging regions ... and ...
2014-09-24 16:38:59,711 INFO  [main] ch18.Merge: Merging regions ... and ...
2014-09-24 16:38:59,713 INFO  [main] ch18.Merge: Merging regions ... and ...

Regions Size

First, you will want to make sure the maximum file size is set to at least 10 GB. It is not an issue to have regions smaller than that, but you want to make sure that when required, regions can get as big as 10 GB. Presplitting is discussed later in this section. It is recommended to monitor region sizes using a visualization tool such as Hannibal. Watch for regions that are growing faster than others, and look to split them during off-peak usage times. Figure 14-3 illustrates how Hannibal helps you to view the size of each region.

Figure 14-3. Visual overview of region sizes courtesy of Hannibal

HBase default split policy (org.apache.hadoop.hbase.regionserver.IncreasingToUpperBoundRegionSplitPolicy) ensures that regions never go above the configured maximum file size (hbase.hregion.max.filesize). However, it might create many smaller regions when cluster load is small (fewer than 100 regions). To make sure your regions can grow to a decent size, validate that hbase.hregion.max.filesize property is configured to at least 10 GB. This property is represented in bytes; therefore it should be 10737418240. However, it is also a common practice to set the maximum file size value to a higher number up to 100 GB. When using larger region sizes, it is important to manage the splits manually. This will make sure the splits occur at the right time and have a minimum impact on the HBase cluster operations and SLAs.

Even if you expect your table to be small, you can still presplit it to increase distribution across RegionServers or manually trigger the splits to make sure you have up to as many regions as you have RegionServers. However, because HBase works better with bigger regions, make sure all your regions are at least 1 GB.

Key and Table Design

The second thing you want to look at is how your application has been designed. One of the most common mistakes in HBase application design is to abuse the number of column families. Because a memstore for a specific region will be divided between the different column families, having too many of them will reduce the size of the memory available for each and might result in very small flushes for some column families. Also, because of the size of the flushed files, it will generate extra compactions. Reducing the number of column families will allow more memory per region for the remaining column families, will reduce the required flushes and compactions, and will allow you to host more regions per server.

Another common mistake on table design is related to presplitting. Based on your key design and the size of the dataset, you will want to make sure that the table is correctly presplit before starting to inject data into the cluster. If the table has not been presplit correctly, you may end up with many unused or underutilized regions. Unused and underutilized regions still require resources such as memory and CPU, and still require periodic compactions. If your data is not evenly distributed but the table is evenly split, you will also end up with some regions being heavily used and some others being almost not used.