Apache HBase: BulkLoad Replication in CDH 6.3 and CDP-DC 7.0

Bulk loading is the process of preparing and loading HFiles (HBase’s own file format) directly into the RegionServers, thus bypassing the write path. 

This obviates many issues, such as:

• MemStores getting full
• WALs getting bigger
• Compaction and flush queues getting swollen
• Garbage Collector getting out of control because of inserts range in the megabytes
• Latency shoots up while importing data

HBase already has a replication functionality called Cluster Replication (see official documentation here). Why isn’t that enough if you want to replicate the bulk loaded data as well?

The “standard” replication uses a source-push methodology. When the active cluster (source) receives an edit to a column family with replication enabled, that edit is propagated to all destination clusters using the WAL for that column family on the RegionServer managing the relevant region.

However, in the case of bulk load, only the event (the fact that there is bulk load happening) is captured in the WAL, with reference to the HFile. The data being loaded is not recorded.

By enabling BulkLoad Replication, the active HBase RegionServer will also send these WAL entries to the peer cluster. Peer cluster will read these WAL entries and copy the HFiles from the active source cluster in the peer cluster staging directory, and basically, from here it’s just a standard bulk load.

For more information on bulk loading, you can refer to a previous blog post or see the Apache documentation. A good function description on bulk loading can be found attached to the HBASE Jira.

Bulk load Replication is supported from CDH 6.3 (including CDP releases).

HFile copy mechanism and setting it up

In order to copy the HFiles from the source cluster to the peer cluster, the peer cluster will need the active cluster’s FileSystem client configurations. FileSystem (FS) could be HDFS or any other file system supported by HBase.

There could be multiple active clusters, so the peer needs to be able to identify the client configurations. Each active cluster must configure a unique replication cluster id. This ID will be sent to the peer cluster as part of a replication request. 

The default implementation (making a custom implementation is out of the scope of this post) for getting the source cluster’s FileSystem client configurations needs each peer cluster to have a replication configuration directory. This can be configured using the hbase.replication.conf.dir property. The directory will have a sub-directory for each one of the active source clusters. Sub-directory name must be the same as the active cluster unique replication id. So the path to the respective FS client configurations will be hbase.replication.conf.dir/hbase.replication.cluster.id.

Note: If there are any changes in FS client configurations in the replication configuration directory, all the RegionServers in the peer cluster must be restarted.

Replication HFile Cleaner

If compaction, merge or split happens, the HFile is moved to the archive folder. This HFile will be cleaned up by HFile cleaner periodically. To avoid this before replication completes, the replication module will maintain these bulk loaded HFile paths into ZooKeeper for each peer. 

• Hbase
  • hfile-refs
    • hfile_name_1
    • hfile_name_2
    • [...]
    • hfile_name_N
    • hfile_name_1
    • hfile_name_2
    • [...]
    • hfile_name_N
  • peer_1
  • peer_m
• replication

The HFile Cleaner will not clean the HFiles until its entry is found in ZooKeeper. Once the HFile is replicated successfully, the ZooKeeper entry will be deleted by the replication module for that particular peer.

Configuring BulkLoad Replication on CDP

Note: If you have Kerberos setup, please check this page on how to set up cross-realm authentication.

Configuring the clusters

1. Using Cloudera Manager, navigate on the source cluster to HBase > Configuration.
   
2. Search for Advanced Configuration Snippet to find the HBase Service Advanced Configuration Snippet (Safety Valve) for hbase-site.xml and add the following configuration properties:
   
   • Name: hbase.replication.bulkload.enabled
     Value: true
   • Name: hbase.replication.cluster.id
     Value: short source cluster ID, for example: source
     
3. Click Save Changes.
   
4. Click the Stale Service Restart icon that is next to the Service to invoke the cluster restart wizard to restart Stale Services.
   
5. Using Cloudera Manager, navigate on the sink cluster to HBase > Configuration.
   
6. Search for Advanced Configuration Snippet again to find the HBase Service Advanced Configuration Snippet (Safety Valve) for hbase-site.xml and add the following configuration property:
   • Name: hbase.replication.conf.dir
     Value: path to the source cluster’s copied configuration directory, for example: /opt/cloudera/fs_conf
     
7. Create the directory on the sink cluster that is defined by the hbase.replication.conf.dir property.
   
8. Copy the source cluster’s HBase client configuration files (core-site.xml, hdfs-site.xml, and hbase-site.xml) to the standby cluster.
9. The settings files should be under /run/cloudera-scm-agent/process/by default, in the master hbase service directory. You can also search for the configuration: