{"id":55302,"date":"2020-01-17T00:00:00","date_gmt":"2020-01-16T21:00:00","guid":{"rendered":"https:\/\/prohoster.info\/blog\/blog_prohoster\/teoriya-i-praktika-ispolzovaniya-hbase"},"modified":"2020-02-18T14:03:24","modified_gmt":"2020-02-18T11:03:24","slug":"teoriya-i-praktika-ispolzovaniya-hbase","status":"publish","type":"post","link":"https:\/\/prohoster.info\/en\/blog\/administrirovanie\/teoriya-i-praktika-ispolzovaniya-hbase","title":{"rendered":"Theory and Practice of Using HBase","gt_translate_keys":[{"key":"rendered","format":"text"}]},"content":{"rendered":"

Good day! My name is Danil Lipovoy, and our team at Sbertech has started using HBase as an operational data store. During its study, we have accumulated experience that we wanted to systematize and describe (we hope many will find it useful). All the experiments below were conducted with versions HBase 1.2.0-cdh5.14.2 and 2.0.0-cdh6.0.0-beta1. <\/p>\n

    \n
  1. Overall Architecture <\/li>\n
  2. Data Writing in HBASE<\/li>\n
  3. Data Reading from HBASE<\/li>\n
  4. Data Caching<\/li>\n
  5. Batch Processing with MultiGet\/MultiPut<\/li>\n
  6. Table Region Splitting Strategy<\/li>\n
  7. Fault Tolerance, Compaction, and Data Locality<\/li>\n
  8. Settings and Performance<\/li>\n
  9. Load Testing<\/li>\n
  10. Conclusions<\/li>\n<\/ol>\n

    <\/a><\/noindex><\/p>\n

    1. Overall Architecture<\/h2>\n

    \n\"Theory
    \nThe backup Master listens for heartbeats from the active one on the ZooKeeper node and takes over the master functions in case of its disappearance. <\/p>\n

    2. Data Writing in HBASE<\/h2>\n

    \nFirst, let\u2019s consider the simplest case \u2013 writing a key-value object to a table using put(rowkey). The client must first determine where the root region server (Root Region Server \u2014 RRS) is located, which holds the table hbase:meta. This information is obtained from ZooKeeper. After that, it contacts the RRS and reads the hbase:meta table, from which it retrieves information regarding which RegionServer (RS) is responsible for storing data for the specified rowkey in the table of interest. For further use, the meta-table is cached by the client, allowing subsequent requests to be faster, directly to the RS.<\/p>\n

    Next, upon receiving the request, the RS first writes it to the WriteAheadLog (WAL), which is necessary for recovery in case of a failure. It then saves the data in MemStore. This is a memory buffer that contains a sorted set of keys for this region. The table can be divided into regions (partitions), each containing a non-overlapping set of keys. This allows for better performance by placing regions on different servers. However, despite the obviousness of this statement, we will see later that it does not work in all cases.<\/p>\n

    After placing the record in MemStore, the client receives a response that the record has been successfully saved. However, it is actually stored only in the buffer and will be written to disk only after a certain interval of time or when it becomes filled with new data. <\/p>\n

    \"Theory
    \nWhen performing the 'Delete' operation, the physical deletion of data does not occur. They are simply marked as deleted, and the actual destruction happens when the major compact function is called, which is described in more detail in section 7.<\/p>\n

    Files in HFile format accumulate in HDFS, and from time to time, a minor compact process is launched, which just merges small files into larger ones without deleting anything. Over time, this becomes a problem that manifests itself only when reading data (we will return to this shortly). <\/p>\n

    In addition to the loading process described above, there is a much more efficient procedure that arguably represents the strongest side of this database \u2013 BulkLoad. This involves us manually creating HFiles and placing them on disk, which allows for excellent scalability and achieving quite impressive speeds. In essence, the limitation here is not HBase, but the capabilities of the hardware. Below are the loading results on a cluster consisting of 16 RegionServers and 16 NodeManager YARN (CPU Xeon E5-2680 v4 @ 2.40GHz * 64 threads), HBase version 1.2.0-cdh5.14.2. <\/p>\n

    \"Theory
    \n
    \nIt can be seen that by increasing the number of partitions (regions) in the table, as well as the Spark executors, we observe an increase in loading speed. The speed also depends on the volume of the write operation. Larger blocks provide gains in MB\/sec, while smaller blocks contribute to the number of inserted records per unit of time, assuming other factors are equal. <\/p>\n

    You can also start loading into two tables simultaneously and achieve double the speed. Below, it can be seen that writing 10 KB blocks to two tables happens at a speed of about 600 Mb\/sec for each (totaling 1275 Mb\/sec), which matches the writing speed to a single table of 623 MB\/sec (see #11 above).<\/p>\n

    \"Theory
    \nThe second run with 50 KB records shows that the loading speed increases only marginally, indicating that we are approaching the threshold values. It should be noted that there is virtually no load on HBASE itself; all that is required from it is to first provide data from hbase:meta, and after the HFiles are placed, to flush the BlockCache and save the MemStore buffer to disk, if it is not empty.<\/p>\n

    3. Reading data from HBASE<\/h2>\n

    \nAssuming that the client already has all the information from hbase:meta (see point 2), the request goes directly to the RS where the required key is stored. First, the search occurs in MemCache. Regardless of whether there is data there or not, the search also occurs in the BlockCache buffer and, if necessary, in HFiles. If the data is found in the file, it is placed in BlockCache and will be returned faster in the next request. The search in HFile is relatively quick due to the use of a Bloom filter, meaning that by reading a small amount of data, it can immediately determine whether this file contains the required key, and if not, it moves on to the next one.<\/p>\n

    \"Theory
    \nAfter obtaining data from these three sources, the RS forms the response. In particular, it may return several found versions of the object if the client requested versioning.<\/p>\n

    4. Data Caching<\/h2>\n

    \nThe MemStore and BlockCache buffers consume up to 80% of the allocated on-heap memory of the RS (the rest is reserved for service tasks of the RS). If the typical usage mode is such that processes write and immediately read the same data, it makes sense to reduce BlockCache and increase MemStore, as when writing, data does not go into the read cache, thus BlockCache will be utilized less frequently. The BlockCache buffer consists of two parts: LruBlockCache (always on-heap) and BucketCache (usually off-heap or on SSD). BucketCache should be used when there are many read requests that do not fit in LruBlockCache, leading to active Garbage Collector operation. However, one should not expect a radical increase in performance from using the read cache, but we will return to this in point 8.<\/p>\n

    \"Theory
    \nBlockCache is shared across the entire RS, while MemStore is specific to each table (one for each Column Family).<\/p>\n

    How is described<\/a><\/noindex> In theory, when writing, data does not go into the cache, and in fact, such parameters as CACHE_DATA_ON_WRITE for the table and 'Cache DATA on Write' for the RS are set to false. However, in practice, if you write data to MemStore, then flush it to disk (effectively clearing it), and then delete the resulting file, executing a get request will still successfully return the data. Moreover, even if BlockCache is completely disabled and the table is filled with new data, then after flushing MemStore to disk, deleting them, and querying from another session, they will still be retrieved from somewhere. Therefore, HBase retains not just data, but also mysterious enigmas.<\/p>\n

    hbase(main):001:0> create 'ns:magic', 'cf'\nCreated table ns:magic\nTook 1.1533 seconds\nhbase(main):002:0> put 'ns:magic', 'key1', 'cf:c', 'try_to_delete_me'\nTook 0.2610 seconds\nhbase(main):003:0> flush 'ns:magic'\nTook 0.6161 seconds\nhdfs dfs -mv \/data\/hbase\/data\/ns\/magic\/* \/tmp\/trash\nhbase(main):002:0> get 'ns:magic', 'key1'\n cf:c      timestamp=1534440690218, value=try_to_delete_me\n<\/code><\/pre>\n
    \nThe parameter 'Cache DATA on Read' is set to false. If you have any ideas, feel free to discuss them in the comments.<\/p>\n
    5. Batch processing of data MultiGet\/MultiPut<\/h2>\n
    \nProcessing single requests (Get\/Put\/Delete) is quite an expensive operation, so it is advisable to combine them whenever possible into a List or List, which allows for significant performance gains. This is especially true for write operations, but there is a catch when it comes to reading. The graph below shows the time taken to read 50,000 entries from MemStore. The reading was performed in a single thread, and the horizontal axis shows the number of keys in the request. It is evident that as the number of keys in a single request increases to a thousand, the execution time decreases, i.e., the speed increases. However, with the default MSLAB mode enabled, after reaching this threshold, performance drastically declines, and the larger the amount of data in the entry, the longer the processing time. <\/p>\n
    \"Theory
    \n
    \nThe tests were conducted on a virtual machine with 8 cores, using HBase version 2.0.0-cdh6.0.0-beta1.<\/p>\n
    The MSLAB mode is designed to reduce heap fragmentation caused by mixing new and old generation data. To address this issue, when MSLAB is enabled, data is placed into relatively small chunks and processed in batches. As a result, when the size of the requested data packet exceeds the allocated size, performance drops sharply. On the other hand, disabling this mode is also undesirable because it leads to stops due to GC during intensive data operations. A good solution is to increase the chunk sizes when actively writing via put while simultaneously reading. It is worth noting that the problem does not occur if the flush command is executed after writing, which flushes the MemStore to disk, or if loading is performed using BulkLoad. The table below shows that requests from MemStore with larger data volumes (and the same number) lead to slowdowns. However, by increasing the chunksize, processing time can be returned to normal.<\/p>\n
    \"Theory
    \nIn addition to increasing chunksize, data partitioning by regions, i.e., table splitting, helps. This results in fewer requests for each region, and if they fit within a cell, the response remains good.<\/p>\n
    6. Table Partitioning Strategy by Regions (Splitting)<\/h2>\n
    \nSince HBase is a key-value store and partitioning is performed by key, it is crucial to evenly distribute data across all regions. For example, partitioning such a table into three parts will result in data being divided into three regions:<\/p>\n
    \"Theory
    \nSometimes, this leads to a significant slowdown if the data being loaded later consists of long values that mostly start with the same digit, for example:<\/p>\n
    1000001
    \n1000002
    \n\u2026
    \n1100003<\/p>\n
    Since keys are stored as byte arrays, all of them will start similarly and belong to region #1, which holds that range of keys. There are several partitioning strategies:<\/p>\n
    HexStringSplit \u2013 Converts the key into a string with hexadecimal encoding in the range \"00000000\" => \"FFFFFFFF\" and fills with zeros on the left.<\/p>\n
    UniformSplit \u2013 Converts the key into a byte array with hexadecimal encoding in the range \"00\" => \"FF\" and fills with zeros on the right.<\/p>\n
    Additionally, you can specify any range or set of keys for partitioning and configure auto-splitting. However, one of the simplest and most effective approaches is UniformSplit, utilizing hash concatenation, for example, the upper pair of bytes from running the key through the CRC32(rowkey) function and the rowkey itself:<\/p>\n
    hash + rowkey<\/p>\n
    Then, all data will be evenly distributed across regions. When reading, the first two bytes are simply discarded, leaving the original key. Moreover, the RS monitors the amount of data and keys in the region, and if limits are exceeded, it automatically splits it into parts. <\/p>\n
    7. Fault Tolerance and Data Locality<\/h2>\n
    \nSince each set of keys is managed by only one region, addressing issues related to RS failures or decommissioning involves storing all necessary data in HDFS. When an RS fails, the master detects this through the absence of a heartbeat on the ZooKeeper node. It then assigns the affected region to another RS, and because HFiles are stored in a distributed file system, the new host reads them and continues serving the data. However, since some data may still be in MemStore and not yet written to HFiles, the WAL is used to restore the operation history, with these logs also stored in HDFS. After applying the changes, the RS is able to respond to requests; however, the relocation results in some data and the processes servicing it being on different nodes, thus decreasing locality. <\/p>\n
    The solution to the problem is major compaction \u2013 this procedure moves files to the nodes responsible for them (where their regions are located), resulting in a significant increase in load on the network and disks during this process. However, this ultimately speeds up data access noticeably. Additionally, major compaction consolidates all HFiles into a single file within the region and cleans up data according to table settings. For instance, one can specify the number of object versions to retain or the time after which the object is physically deleted.<\/p>\n
    This procedure can have a very positive impact on HBase performance. The picture below shows how performance degraded as a result of intensive data writing. Here, it's evident that 40 threads were writing to a single table while 40 other threads simultaneously read data. The writing threads generated more and more HFiles, which were read by the other threads. Consequently, more data needed to be removed from memory and eventually, GC kicked in, which practically paralyzed all operations. Initiating major compaction helped clean up the generated debris and restore performance.<\/p>\n
    \"Theory
    \nThe test was conducted on 3 DataNodes and 4 RS (CPU Xeon E5-2680 v4 @ 2.40GHz * 64 threads). HBase version 1.2.0-cdh5.14.2<\/p>\n
    It is worth noting that the major compaction was performed on a \"live\" table, where data was actively being written and read. There were claims in the network that this could lead to incorrect responses when reading data. To verify this, a process was initiated that generated new data and wrote it to the table. Immediately afterwards, it read the data and checked if the retrieved value matched what had been written. During the operation of this process, major compaction was initiated about 200 times without any failures recorded. The problem may occur rarely and only under high load, so it is safer to schedule the stopping of read and write processes and perform the cleanup to avoid such GC drops.<\/p>\n
    Moreover, major compaction does not affect the state of MemStore, to flush it to disk and compact it, flush must be used (connection.getAdmin().flush(TableName.valueOf(tblName))).<\/p>\n
    8. Settings and Performance<\/h2>\n
    \nAs mentioned earlier, HBase shows the greatest success where it has nothing to do during BulkLoad. However, this applies to most systems and people. Yet, this tool is primarily suited for bulk data loading, while if the process requires performing many competing read and write requests, the aforementioned Get and Put commands are used. To determine the optimal parameters, runs were conducted with different combinations of table parameters and settings:<\/p>\n