HBase Pros and Cons

5. HBase Physical Architecture

Figure 5.1 shows the deployment view for HBase cluster:
HBase is the master-slave cluster on top of HDFS. The classic deployment is as follows:
➢ Master node: one HMaster and one NameNode running on a machine as the master node.
➢ Slave node: Each node is running one HRegionServer and one DataNode. And each node report status to the master node and Zookeeper.
[more…]
➢ Zookeeper: HBase is shipped with ensemble Zookeeper, but for large clusters, using existing Zookeeper is better. Zookeeper is crucial, the HMaster and HRegionServers will register on Zookeeper.
➢ Client: There can be many clients to access HRegionServer, like Java Client, Shell Client, Thrift Client and Avro Client

Figure 5.1 HBase Deployment View

6.HBase Pros

1. Master-Slave Architecture
HBase build on top HDFS, HDFS is mater-slave architecture, HBase follows this style. It will improve the interoperability between HBase and HDFS.
This style do good for load balance, the HMaster takes over the load balance job, assign regions to region servers and auto failover dead RegionServers.

2. Real-time , random big data access
HDFS handles large blocks of data well, but small blocks of data with real-time access is not efficient. HBase uses LSM-trees as data storage architecture. Data is written into WAL first, then to Mem Store. WAL is in case of data lost before Mem Store is flushed to HDFS. Mem Store will be flushed to HDFS when it’s filled up. The mechanism insures that data write is operated in memory, no need of HDFS disk access, which improves write performance.
Meanwhile, data read accesses Mem Store first, then to Block Cache, and last the HFiles, so for random data access, if there is cache in memory, the read performance is very well.
In short, HBase brings big data online, it is efficient for real-time operations on big data(TB, PB);

3. Column-Oriented data model for big sparse table
HBase is NO-SQL, column-family-oriented, each column family is stored together in HStore. The key-value data model stores only non-empty values, no null values. So for large spares table, the disk usage is great in HBase, not like RDBMS storing large quantities of null values.

4. LSM-trees vs. B+ tree
HBase data storage architecture is LSM-trees. A big feature of LSM-trees is the merge housekeeping, which merge small files into a larger one to reduce disk seek. The housekeeping is a background process, which will not impact real-time data access.
LSM-trees transform random data access into sequential data access, which improves read performance. But B+ tree in RDBMS requires more disk seeks for data modifications, which is very efficient for big data process.

5. Row-level Atomic
HBase has no strictly ACID features like RDBMS, how to insure data consistent in HBase and how to avoid dirty read. HBase is row-level automic, in other words, each Put operation on a given row either success or fail. Meanwhile, CheckAnd*, Increment* operations are also atomic.
The scan operation across the table is not on a snapshot of a table, if the data is updated before a Scan operation, the updated version is read by Scan, It ensures data consistent.
HBase is good at loose coupled data and not high requirements for consistent. Row-level atomic is a mechanism to improve data consistent, but data consistent is really a shortage of HBase.

6. High Scalability
HBase data is stored across the cluster, the cluster has many HRegionServer, which can be scaled. HBase build on top of HDFS, HDFS is high scalability, that is the DataNode can be scaled. HDFS brings great scalability to HBase. Moreover, the master-slave architecture do well in data scalability.

7. High Reliability
The HBase cluster is built on commodity hardware, which will be dead at any time. HBase data is replicated across the cluster. Data replication is done by HDFS. HDFS stores 3 replications for each block: on the same node, on another node on the same rack, on another node on another rack. The feature brings high reliability for HBase.

8. Auto Failover
HMaster is in charge of auto failover work. When HMaster detects the dead HRegionServer. It will find that the region assignment is invalid, and it will do region assignment again. Auto failover brings great availability for HBase, which is no need of manually HRegionServer failover.

9. Data auto sharding
Since data is stored distributed across the cluster. HBase must makes use of all of the machines in the cluster. HRegionServer stores many HRegions. When a HRegion size reaches its threshold, HRegionServer will split it into half.
In addition, when the number of regions is over the threshold that the HMaster can handle, the HMaster will merge small regions into a larger one. Auto sharding brings HBase great load balance and keeps programmer away from the details of distributed data sharding.

10. Simple Client Interface
HBase provides 5 basic data operations : Put, Get, Delete, Scan and Increment. The Interface is easy to use. HBase provide java interface, and non java interface: Thrift, Avro, REST and Shell. Each interface is transparent for developers to do data operations on HBase.

7. HBase Cons

1. Single Point of Failure (SPOF)
The master-slave architecture always has SPOF, HBase is no exception. When there is only on HMaster in a cluster, if the cluster has more than 4000 nodes, the HMaster is the performance bottleneck. Even though client talks directly to HRegionServer, when the HMaster is down, the cluster can be “steady” for a short time, HMaster long time down will bring down the cluster.
It is possible to start multi HMaster in a cluster, but only one is active, which is also a performance bottleneck.

2. No transaction
As we talked before, HBase data consistent is a shortage. Interrow operations are not atomic, which will bring dirty read. HBase is good at louse coupled data but not good at highly rational data records. That is an important point we must be aware of.

3. No Join, if you want, use MapReduce
HBase has no join operation, so data model has to be redesign when migrate from RDBMS, which is a big headache. For highly rational data records, HBase will make you sick. But if you really want implement join, you can use MapReduce, please refer to “HBase in Action”.

4. Index only on key, sorted by key, but RDBMS can be indexed on arbitrary column
The column-oriented data model is key-value style, HBase only has index on key: <rowkey, column family, qualifier, version>, rowkey design is the most important thing.
The data model paradigm for HBase is quite different for RDBMS. HBase has no index on value. Scan with columnvalue filter for big data is very slow. But RDBMS can be indexed on arbitrary column.
If you want to improve index for HBase, you can use MapReduce to build Inverted Index.

5. Security Problem
Finally, HBase has security problem. RDBMS like MySQL has authentication and authorization feature, different user has different data access authority. But HBase has no such feature yet. But we can see that, improve security will lower the performance, If the online application can insure the security, HBase will has no need to care about that.

Refereces
[1] George, Lars. HBase: the definitive guide. O’Reilly Media, Inc., 2011.
[2] Dimiduk, Nick, Amandeep Khurana, and Mark Henry Ryan. HBase in action. Manning, 2013.
[3] Chang, Fay, et al. “Bigtable: A distributed storage system for structured data.”ACM Transactions on Computer Systems (TOCS) 26.2 (2008): 4.
[4] White, Tom. Hadoop: The Definitive Guide: The Definitive Guide. O’Reilly Media, 2009.
[5] Kruchten, Philippe B. “The 4+ 1 view model of architecture.” Software, IEEE12.6 (1995): 42-50.
[6] Apache HBase, “The Apache HBase Reference Guide”, apache.org, 2013

From: http://www.cyanny.com/2014/03/13/hbase-architecture-analysis-part-3-pros-cons/

HBase Process Analysis

4. Process Architecture

4.1 HBase Write Path

The client doesn’t write data directly into HFile on HDFS. Firstly it writes data to WAL(Write Ahead Log), and Secondly, writes to MemStore shared by a HStore in memory.
[more…]

Figure4.1 HBase Write Path

MemStore is a write buffer(64MB by default). When the data in MemStore accumulates its threshold, data will be flush to a new HFile on HDFS persistently. Each Column Family can have many HFiles, but each HFile only belongs to one Column Family.
WAL is for data reliability, WAL is persistent on HDFS and each Region Server has only on WAL. When the Region Server is down before MemStore flush, HBase can replay WAL to restore data on a new Region Server.
A data write completes successfully only after the data is written to WAL and MemStore.

4.2 HBase Read Path

As shown in Figure 4.2, it’s the read path of HBase.
1. Client will query the MemStore in memory, if it has the target row.
2. When MemStore query failed, client will hit the BlockCache.
3. After the MemStore and BlockCache query failed, HBase will load HFiles into memory which may contain the target row info.
The MemStore and BlockCache is the mechanism for real time data access for distributed large data.
BlockCache is a LRU(Lease Recently Used) priority cache. Each RegionServer has a single BlockCache. It keeps frequently accessed data from HFile in memory to reduce disk data reads. The “Block”(64KB by default) is the smallest index unit of data or the smallest unit of data that can be read from disk by one pass.
For random data access, small block size is preferred, but block index consumes more memory. And for sequential data access, large block size is better, fewer index save more memory.

Figure 4.2 HBase Read Path

4.3.HBase Housekeeping: HFile Compaction

The data of each column family is flush into multiple HFiles. Too many HFiles means many disk data reads and lower the read performance. Therefore, HBase do HFile compaction periodically.

Figure4.3 HFile Compaction

➢Minor Compaction
It happens on multiple HFiles in one HStore. Minor compaction will pick up a couple of adjacent small HFiles and rewrite them into a larger one.
The process will keep the deleted or expired cells. The HFile selection standard is configurable. Since minor compaction will affect HBase performace, there is an upper limit on the number of HFiles involved (10 by default).
➢Major Compaction
Major Compaction compact all HFiles in a HStore(Column Family) into one HFile. It is the only chance to delete records permanently. Major Compaction will usually have to be triggered manually for large clusters.
Major Compaction is not region merge, it happens to HStore which will not result in region merge.

4.4 HBase Delete

When HBase client send delete request, the record will be marked “tombstone”, it is a “predicate deletion”, which is supported by LSM-tree. Since HFile is immutable, deletion isn’t available for HFile on HDFS. Therefore, HBase adopts major compaction(Section 4.3) to clean up deleted or expired records.

4.5 Region Assignment

It is a main task of HMaster:
1. HMaster invoke the AssignmentManager to do region assignment.
2. AssignmentManager checks the existing region assignments in .META.
3. If region assignment is valid, then keep the region.
4. If region assignment is invalid, then the LoadBalancerFactory will create a DefaultLoadBalancer
5. DefaultLoadBalancer will assign the new region randomly to a RegionServer
6. Update the assignment to .META.
7. The RegionServer open the new region

4.6 Region Split

Region split is the work of RegionServer, not participated by HMaster. When a region in a RegionServer accumulates over size threshold, RegionServer will split the region into half.
1. RegionServer offline the region to be split.
2. RegionServer split the region into two half
3. Update new daughter regions info to .META.
4. Open the new daughter regions.
5. Report the split info to HMaster.

4.7 Region Merge

A RegionServer can’t have too many regions, because too many regions bring large cost of memory and lower the performance of RegionServer. Meanwhile, HMaster can’t handle load balance with too many regions.
Therefore, when the number of regions is over a threshold, region merge is trigged. Region Merge is joined by RegionServer and HMaster.
1. Client send RPC region merge request to HMaster
2. HMaster moves regions together to the same RegionServer where the more heavily loaded region resided.
3. HMaster send request to the RegionServer to run region merge.
4. The RegionServer offline the regions to be merged.
5. The RegionServer Merge the regions on the local file system.
6. Delete the meta info of the merging regions on .META. table, add new meta info to .META. table.
7. Open the new merged region
8. Report the merge to HMaster

4.8 Auto Failover

4.8.1 RegionServer Failover

When a region server fails, HMaster will do failover automatically.
1. RegionServer is down
2. HMaster will detect the unavailable RegionServer when there is no heartbeat report.
3. HMaster start region reassignment, detect that the region assignment is invalid, then re-assign the regions like the region assignment sequence.

4.8.2 Master Failover

In centralized architecture style, the SPOF is a problem, you may ask How does HBase will do when HMaster failed? There two safeguards for SPOF:
1. Multi-Master environment
HBase can run in multi-master cluster. There is only one active master, when the master is down, the remaining master will take over the master role.
It looks like Hadoop HDFS new feature high availability, a standby NameNode will take over the NameNode role when the active one failed.
2. Catalog tables are on RegionServers
When client send read/write request to HBase, it talks directly to RegionServer, not HMaster. Hence, the cluster can be steady for a time without HMaster. But it requires that the HMaster failover as soon as possible, HMaster is the vital of the cluster.

4.9 Data Replication

For data reliability, HBase replicates data blocks across the cluster. This is achieved by Hadoop HDFS.
By default, there 3 replicas:
1. The First replica is written to local node.
2. The second replica is written to a random node on a different rack.
3. The third replica is on a random node on the same rack
HBase stores HFiles and WAL on HDFS. When RegionServer is down, new region assignment can be done by read replicas of HFile and replay the replicas the WAL. In short, HBase has high reliability.

From: http://www.cyanny.com/2014/03/13/hbase-architecture-analysis-part2-process-architecture/

HBase Architecture Analysis

1. Overview

Apache HBase is an open source column-oriented database. It is often described as a sparse, consistent, distributed, multi-dimensional sorted map. HBase is modeled after Google’s “Bigtable: A distributed Storage System for Structured Data”, which can host very large tables with billions of rows, X millions of columns.
[more…]
HBase is a No-SQL database, and it has very different paradigms from traditional RDBMS, which speaks SQL and enforce relationships upon data. HBase stores structured and semi-structured data with key-value style. Each row in HBase is located by <Rowkey, Column Family, Column Qualifier, Version>.

HBase stores data distributed on Hadoop HDFS shipped with random data access. HDFS is Hadoop Distributed File System, which stores big data sets with streaming style. It has high reliability, scalability and availability but the cons is that HDFS split data into blocks (64MB or 128MB), it can handle large chucks of data very well, but not very well for small piece of data and low-latency data access. In other words, HDFS isn’t fit for online real time read and write, and that’s the meaning of HBase. HBase goal is to make use of cloud distributed data storage for online real time data access.
The blog analyzes the architecture of HBase with the method of “4+1” views by Kruthten.

2. Use Case View

Figure 2.1 User Case View of HBase
The figure 2.1 shows the use case view of HBase. Since from the architecture view, the figure gives more details about the main features of HBase:
As shown on figure 2.1, the main features of HBase are as follows:
1. For User:

•  Read/Write big data in real time: HBase provides transparent and simple interface for user. User has no need to care about details about data transforming.
• Manage Data Model: User Need to create and manage column-oriented data model about business logic, and HBase provide Interface for them to do management.
• Manage System: User can manually configure and monitor the status of HBase

2. For the System

• Store data distributed on HDFS: HBase has great scalability and reliability based on HDFS.
•  Data random online, real time access: HBase provides block cache and index on files for real time queries.
• Automatic Data Sharding: HBase partitions big data automatically on distributed node.
•  Automatic failover: Since hardware failure on clusters is inevitable, HBase can failover automatically for high reliability and availability.

3. Logical Architecture

3.1 High Level Architecture and Style

Figure 3.1 High Level Architecture for HBase
As shown in Figure3.1, the HBase architecture is layered overall, layered architecture style is a typical style for database.
1. Layered Architecture Style
➢ Client Layer: The layer is for user, HBase provides easy-to-use java client API, and ships with non-java API as well, such as shell, Thrift, Avro and REST.
➢ HBase Layer: The layer is the core logic of HBase. Overall, HBase has two important components: HMaster and RegionServer. We will discuss them later.
Moreover, another important component is Zookeeper, which is a distributed coordination service modeled after Google’s Chubby. It’s another open source project under Apache.
➢ Hadoop HDFS Layer: HBase build on top of HDFS, which provides distributed data storage and data replication.

2. Master-Slave Architecture Style
For distributed applications, how to handle data blocks is a key point. There are two typical styles: centralized and non-centralized. Centralized mode is very available for scalability and load balancer, but there is Single Pont of Failure(SPOF) problem and performance bottleneck on the master node. Non-centralized mode has no SPOF problem and performance bottleneck, but not very easy to do load balance and there is data consistent problem.
HBase utilizes Centralized Mode: Master-Slave Style. BTW, Hadoop is also master-slave. The architecture style in Hadoop ecosystem is kind of consistent master-slave style.

Figure 3.2 Master-Slave Architecture Style of HBase
As shown in figure 3.2:

• HMaster is the master in such style, which is responsible for RegionServer monitor, region assignment, metadata operations, RegionServer Failover etc. In a distributed cluster, HMaster runs on HDFS NameNode.
• RegionServer is the slave, which is responsible for serving and managing regions. In a distributed cluster, it runs on HDFS DataNode.
• Zookeeper will track the status of Region Server, where the root table is hosted. Since HBase 0.90.x, it introduces an even more tighter integration with Zookeeper. The heartbeat report from Region Server to HMaster is moved to Zookeeper, that is zookeeper has the responsibility of tracking Region Server status. Moreover, Zookeeper is the entry point of client, which enable query Zookeeper about the location of the region hosting the –ROOT- table.

3.2 HBase Data Model

HBase data model is Column-Family-Oriented. Some special characteristics of HBase data model is as follows:
➢ Key-Value store: Each cell in a table is specified by the four dimensional key: . The table will not store null values like RDBMS, it works well for large sparse table.
➢ Sorted by key, index only on key. Row key design is the only most important thing in HBase Schema Design.
➢ Each cell can store different versions, specified by Timestamp by default
➢ The column qualifier is schema-less, can be changed during run-time
➢ The column qualifier and value is treated as arbitrary bytes

Table 3.1 The logical view of data model
As shown in Table 3.1, is a logical view of a big table, the table is parse, each row has a unique row key, and the table has three column family: cf1, cf2, cf3. cf1 has two qualifiers: q1-1, q1-2. q1-1 on timestamp t1 has value v1.

Table 3.2 The physical view of data model
In RDBMS, a big table as shown in Table 3.1 will store a lot of null values, but for HBase is not, key-value styles make HBase store only non-empty values as shown in Table3.2. It’s the physical view, a table is partitioned by Column Family, each Column Family is stored in a HStore in a table region on a Region Server.
Since the main purpose of the report is architecture analysis, the table design of HBase will not discuss in detail, you can refer to the “HBase in Action”

3.3 HBase Distributed Mode

3.3.1 Big Table Splitting

HBase can run on local file system, called “standalone mode” for development and testing. For production, HBase should run on cluster on top of HDFS. Big tables are split and stored across the cluster.

Figure 3.3 Region split on big table
A table is split into roughly equal size. Regions are assigned to Region Servers across the cluster. And Region Servers host roughly equal number of regions. As shown in Figure3.3, Table A is split into four regions, each of which is assigned to a Region Server.

3.3.2 Region Server

As shown in figure3.4, the Region Server Architecture. It contains several components as follows:

• One Block Cache, which is a LRU priority cache for data reading.
• One WAL(Write Ahead Log): HBase use Log-Structured-Merge-Tree(LSM tree) to process data writing. Each data update or delete will be write to WAL first, and then write to MemStore. WAL is persisted on HDFS.
•  Multiple HRegions: each HRegion is a partition of table as we talk about in 3.3.1.
• In a HRegion: Multiple HStore: Each HStore is correspond to a Column Family
• In a HStore:  One MemStore: store updates or deletes before flush to disk. Multiple StoreFile, each of which is correspond to a HFile
• A HFile is immutable, flushed from MemStore, persisted on HDFS

Figure3.4 Region Server Architecture

3.3.3 -ROOT- and .META table

Since table are partitioned and store across the cluster. How can the client find which region hosting a specific row key range? There are two special catalog tables, -ROOT- and .META. table for this.
➢.META. table: host the region location info for a specific row key range. The table is stored on Region Servers, which can be split into as many region as required.
➢-ROOT- table: host the .META. table info. There is only one Region Server store the –ROOT- table. And the Root region never split into more than one region.
The –ROOT- and .META. table structure logically looks as a B+ tree as shown in figure 3.5.
The RegionServer RS1 host the –ROOT- table, the .META. table is split into 3 regions: M1, M2, M3, hosted on RS2, RS3, RS1. Table T1 contains three regions, T2 contains four regions. For example, T1R1 is hosted on RS3, the meta info is hosted on M1.

Figure 3.5 –ROOT-, .META., and User table viewed as B+ tree

3.3.4 Region Lookup

How can client find where the –ROOT- table is? Zookeeper does this work, and how to find the region where the target row is. Even though it belongs to process architecture, it is related to the section, hence we discuss it in advance. let’s look at an example as shown in figure3.6.
1. Client query Zookeeper: where is the –ROOT-? On RS1.
2. Client request RS1: Which meta region contains row: T10006? META1 on RS2
3. Client request RS2: Which region can find the row T10006? Region on RS3
4. Client get the from the region on RS3
5. Client cache the region info, and is refreshed until the region location info changed.

Figure 3.6 Region Lookup Process

3.4 HBase Communication Protocol

Figure 3.7 HBase Communication Protocol
In HBase 0.96 release, HBase has moved its communication protocol to Protocol Buffer.Protocol buffer is a method of serializing structured data developed by Google. Google uses it for almost all of its internal RPC protocol and file formats.
Protocol buffer involves an Interface Description Language for cross language service like Apache Thrift.
Basically, the communication between sub-systems of HBase is RPC, which is implemented in Protocol Buffer.
The main protocols are as follows:
➢ MasterMonitor Protocol: client use it to monitor the status of HMaster
➢ MasterAdmin Protocol: client use it to do management for HMaster, such as region manually management, table meta info management.
➢ Admin Protocol: client use it communicate with HRegionServer, to do admin work, such as region split, store file compact, WAL management.
➢ Client Protocol: Client use it to read/write to HRegionServer
➢ RegionServerStatus Protocol: HRegionServer use it to communicate with HMaster: including the request and response of server startup, server fatal error, server status report.

3.5 Log-Structured Merge-Trees(LSM-trees)

No-SQL database usually uses LSM-trees as data storage process architecture. HBase is no exception. As we all known, RDBMS adopts B+ tree to organize its indexes, as shown in Fugure 3.3. These B+ trees are often 3-level n-way balance trees. The nodes of a B+ tree are blocks on disk. So for a update by RDBMS, it likely needs 5 times disk operation. (3 times for B+ tree to find the block of the target row, 1 time for target block read, and 1 time for data update).
On RDBMS, data is written randomly as heap file on disk, but random data block decrease read performance. That’s why we need B+ tree index. B+ tree is fit well for data read, but is not efficient for data updates. Given the large distributed data, B+ tree is not the competitor for LSM-trees so far.

Figure3.8 B+ tree

LSM-trees can be viewed as n-level merge-trees. It transforms random writes into sequential writes using logfile and in-memory store. Figure3.9 shows data write process of LSM-trees.

Figure 3.9 LSM-trees

➢ Data Write(Insert, update): Data is written to logfile sequentially first, then to in-memory store, where data is organized as sorted tree, like B+ tree. When the in-memory store is filled up, the tree in the memory will be flushed to a store file on disk. The store files on disk is arranged like B+ tree, as the C1 Tree shown in Figure 3.9 . But store files are optimized for sequential disk access.
➢Data Read: In-memory store is searched first. Then search the store files on disk.
➢Data Delete: Give a data record a “delete marker”, system background will do housekeeping work by merging some store files into a larger one to reduce disk seeks. A data record will be deleted permanently during the housekeeping.
LSM-trees’ data updates are operated in memory, no disk access, it’s faster than B+ tree. When the data read is always on the data set that is written recently, LSM-trees will reduce disk seeks, and improve performance. When disk IO is the cost we must consider, LSM-trees is more suitable than B+ tree.

From: http://www.cyanny.com/2014/03/13/hbase-architecture-analysis-part1-logical-architecture/

Distributed locks with Redis

Distributed locks are a very useful primitive in many environments where different processes require to operate with shared resources in a mutually exclusive way.

There are a number of libraries and blog posts describing how to implement a DLM (Distributed Lock Manager) with Redis, but every library uses a different approach, and many use a simple approach with lower guarantees compared to what can be achieved with slightly more complex designs.

This page is an attempt to provide a more canonical algorithm to implement distributed locks with Redis. We propose an algorithm, called Redlock, which implements a DLM which we believe to be safer than the vanilla single instance approach. We hope that the community will analyze it, provide feedback, and use it as a starting point for the implementations or more complex or alternative designs.

Implementations

Before describing the algorithm, here there are a few links at implementations already available, that can be used as a reference.

• Redlock-rb (Ruby implementation). There is also a fork of Redlock-rb that adds a gem for easy distribution and perhaps more.
• Redlock-py (Python implementation).
• Redlock-php (PHP implementation).
• Redsync.go (Go implementation).
• Redisson (Java implementation).
• Redis::DistLock (Perl implementation).
• Redlock-cpp (Cpp implementation).
• Redlock-cs (C#/.NET implementation).
• node-redlock (NodeJS implementation). Includes support for lock extension.

Safety and Liveness guarantees

We are going to model our design with just three properties, that are from our point of view the minimum guarantees needed to use distributed locks in an effective way.

1. Safety property: Mutual exclusion. At any given moment, only one client can hold a lock.
2. Liveness property A: Deadlocks free. Eventually it is always possible to acquire a lock, even if the client that locked a resource crashed or gets partitioned.
3. Liveness property B: Fault tolerance. As long as the majority of Redis nodes are up, clients are able to acquire and release locks.

Why failover based implementations are not enough

To understand what we want to improve, let’s analyze the current state of affairs with most Redis-based distributed locks libraries.

The simple way to use Redis to lock a resource is to create a key into an instance. The key is usually created with a limited time to live, using Redis expires feature, so that eventually it gets released one way or the other (property 2 in our list). When the client needs to release the resource, it deletes the key.

Superficially this works well, but there is a problem: this is a single point of failure in our architecture. What happens if the Redis master goes down? Well, let’s add a slave! And use it if the master is unavailable. This is unfortunately not viable. By doing so we can’t implement our safety property of the mutual exclusion, because Redis replication is asynchronous.

This is an obvious race condition with this model:

1. Client A acquires the lock into the master.
2. The master crashes before the write to the key is transmitted to the slave.
3. The slave gets promoted to master.
4. Client B acquires the lock to the same resource A already holds a lock for. SAFETY VIOLATION!

Sometimes it is perfectly fine that under special circumstances, like during a failure, multiple clients can hold the lock at the same time. If this is the case, you can use your replication based solution. Otherwise we suggest to implement the solution described in this document.

Correct implementation with a single instance

Before trying to overcome the limitation of the single instance setup described above, let’s check how to do it correctly in this simple case, since this is actually a viable solution in applications where a race condition from time to time is acceptable, and because locking into a single instance is the foundation we’ll use for the distributed algorithm described here.

To acquire the lock, the way to go is the following:

    SET resource_name my_random_value NX PX 30000

The command will set the key only if it does not already exist (NX option), with an expire of 30000 milliseconds (PX option). The key is set to a value “myrandomvalue”. This value requires to be unique across all the clients and all the locks requests.

Basically the random value is used in order to release the lock in a safe way, with a script that tells Redis: remove the key only if exists and the value stored at the key is exactly the one I expect to be. This is accomplished by the following Lua script:

if redis.call("get",KEYS[1]) == ARGV[1] then
    return redis.call("del",KEYS[1])
else
    return 0
end

This is important in order to avoid removing a lock that was created by another client. For example a client may acquire the lock, get blocked into some operation for longer than the lock validity time (the time at which the key will expire), and later remove the lock, that was already acquired by some other client. Using just DEL is not safe as a client may remove the lock of another client. With the above script instead every lock is “signed” with a random string, so the lock will be removed only if it is still the one that was set by the client trying to remove it.

What this random string should be? I assume it’s 20 bytes from /dev/urandom, but you can find cheaper ways to make it unique enough for your tasks. For example a safe pick is to seed RC4 with /dev/urandom, and generate a pseudo random stream from that. A simpler solution is to use a combination of unix time with microseconds resolution, concatenating it with a client ID, it is not as safe, but probably up to the task in most environments.

The time we use as the key time to live, is called the “lock validity time”. It is both the auto release time, and the time the client has in order to perform the operation required before another client may be able to acquire the lock again, without technically violating the mutual exclusion guarantee, which is only limited to a given window of time from the moment the lock is acquired.

So now we have a good way to acquire and release the lock. The system, reasoning about a non-distributed system which is composed of a single instance, always available, is safe. Let’s extend the concept to a distributed system where we don’t have such guarantees.

The Redlock algorithm

In the distributed version of the algorithm we assume to have N Redis masters. Those nodes are totally independent, so we don’t use replication or any other implicit coordination system. We already described how to acquire and release the lock safely in a single instance. We give for granted that the algorithm will use this method to acquire and release the lock in a single instance. In our examples we set N=5, which is a reasonable value, so we need to run 5 Redis masters in different computers or virtual machines in order to ensure that they’ll fail in a mostly independent way.

In order to acquire the lock, the client performs the following operations:

1. It gets the current time in milliseconds.
2. It tries to acquire the lock in all the N instances sequentially, using the same key name and random value in all the instances. During the step 2, when setting the lock in each instance, the client uses a timeout which is small compared to the total lock auto-release time in order to acquire it. For example if the auto-release time is 10 seconds, the timeout could be in the ~ 5-50 milliseconds range. This prevents the client to remain blocked for a long time trying to talk with a Redis node which is down: if an instance is not available, we should try to talk with the next instance ASAP.
3. The client computes how much time elapsed in order to acquire the lock, by subtracting to the current time the timestamp obtained in step 1. If and only if the client was able to acquire the lock in the majority of the instances (at least 3), and the total time elapsed to acquire the lock is less than lock validity time, the lock is considered to be acquired.
4. If the lock was acquired, its validity time is considered to be the initial validity time minus the time elapsed, as computed in step 3.
5. If the client failed to acquire the lock for some reason (either it was not able to lock N/2+1 instances or the validity time is negative), it will try to unlock all the instances (even the instances it believe it was not able to lock).

Is the algorithm asynchronous?

The algorithm relies on the assumption that while there is no synchronized clock across the processes, still the local time in every process flows approximately at the same rate, with an error which is small compared to the auto-release time of the lock. This assumption closely resembles a real-world computer: every computer has a local clock and we can usually rely on different computers to have a clock drift which is small.

At this point we need to better specify our mutual exclusion rule: it is guaranteed only as long as the client holding the lock will terminate its work within the lock validity time (as obtained in step 3), minus some time (just a few milliseconds in order to compensate for clock drift between processes).

For more information about similar systems requiring a bound clock drift, this paper is an interesting reference:Leases: an efficient fault-tolerant mechanism for distributed file cache consistency.

Retry on failure

When a client is not able to acquire the lock, it should try again after a random delay in order to try to desynchronize multiple clients trying to acquire the lock, for the same resource, at the same time (this may result in a split brain condition where nobody wins). Also the faster a client will try to acquire the lock in the majority of Redis instances, the less window for a split brain condition (and the need for a retry), so ideally the client should try to send the SET commands to the N instances at the same time using multiplexing.

It is worth to stress how important is for the clients that failed to acquire the majority of locks, to release the (partially) acquired locks ASAP, so that there is no need to wait for keys expiry in order for the lock to be acquired again (however if a network partition happens and the client is no longer able to communicate with the Redis instances, there is to pay an availability penalty and wait for the expires).

Releasing the lock

Releasing the lock is simple and involves just to release the lock in all the instances, regardless of the fact the client believe it was able to successfully lock a given instance.

Safety arguments

Is the algorithm safe? We can try to understand what happens in different scenarios.

To start let’s assume that a client is able to acquire the lock in the majority of instances. All the instances will contain a key with the same time to live. However the key was set at different times, so the keys will also expire at different times. However if the first key was set at worst at time T1 (the time we sample before contacting the first server) and the last key was set at worst at time T2 (the time we obtained the reply from the last server), we are sure that the first key to expire in the set will exist for at least MIN_VALIDITY=TTL-(T2-T1)-CLOCK_DRIFT. All the other keys will expire later, so we are sure that the keys will be simultaneously set for at least this time.

During the time the majority of keys are set, another client will not be able to acquire the lock, since N/2+1 SET NX operations can’t succeed if N/2+1 keys already exist. So if a lock was acquired, it is not possible to re-acquire it at the same time (violating the mutual exclusion property).

However we want to also make sure that multiple clients trying to acquire the lock at the same time can’t simultaneously succeed.

If a client locked the majority of instances using a time near, or greater, than the lock maximum validity time (the TTL we use for SET basically), it will consider the lock invalid and will unlock the instances, so we only need to consider the case where a client was able to lock the majority of instances in a time which is less than the validity time. In this case for the argument already expressed above, for MIN_VALIDITY no client should be able to re-acquire the lock. So multiple clients will be able to lock N/2+1 instances at the same time (with “time" being the end of Step 2) only when the time to lock the majority was greater than the TTL time, making the lock invalid.

Are you able to provide a formal proof of safety, point out to existing algorithms that are similar enough, or to find a bug? That would be very appreciated.

Liveness arguments

The system liveness is based on three main features:

1. The auto release of the lock (since keys expire): eventually keys are available again to be locked.
2. The fact that clients, usually, will cooperate removing the locks when the lock was not acquired, or when the lock was acquired and the work terminated, making it likely that we don’t have to wait for keys to expire to re-acquire the lock.
3. The fact that when a client needs to retry a lock, it waits a time which is comparable greater to the time needed to acquire the majority of locks, in order to probabilistically make split brain conditions during resource contention unlikely.

However we pay an availability penalty equal to “TTL” time on network partitions, so if there are continuous partitions, we can pay this penalty indefinitely. This happens every time a client acquires a lock and gets partitioned away before being able to remove the lock.

Basically if there are infinite continuous network partitions, the system may become not available for an infinite amount of time.

Performance, crash-recovery and fsync

Many users using Redis as a lock server need high performance in terms of both latency to acquire and release a lock, and number of acquire / release operations that it is possible to perform per second. In order to meet this requirement, the strategy to talk with the N Redis servers to reduce latency is definitely multiplexing (or poor’s man multiplexing, which is, putting the socket in non-blocking mode, send all the commands, and read all the commands later, assuming that the RTT between the client and each instance is similar).

However there is another consideration to do about persistence if we want to target a crash-recovery system model.

Basically to see the problem here, let’s assume we configure Redis without persistence at all. A client acquires the lock in 3 of 5 instances. One of the instances where the client was able to acquire the lock is restarted, at this point there are again 3 instances that we can lock for the same resource, and another client can lock it again, violating the safety property of exclusivity of lock.

If we enable AOF persistence, things will improve quite a bit. For example we can upgrade a server by sending SHUTDOWN and restarting it. Because Redis expires are semantically implemented so that virtually the time still elapses when the server is off, all our requirements are fine. However everything is fine as long as it is a clean shutdown. What about a power outage? If Redis is configured, as by default, to fsync on disk every second, it is possible that after a restart our key is missing. In theory, if we want to guarantee the lock safety in the face of any kind of instance restart, we need to enable fsync=always in the persistence setting. This in turn will totally ruin performances to the same level of CP systems that are traditionally used to implement distributed locks in a safe way.

However things are better than what they look like at a first glance. Basically the algorithm safety is retained as long as when an instance restarts after a crash, it no longer participates to any currently active lock, so that the set of currently active locks when the instance restarts, were all obtained by locking instances other than the one which is rejoining the system.

To guarantee this we just need to make an instance, after a crash, unavailable for at least a bit more than the max TTL we use, which is, the time needed for all the keys about the locks that existed when the instance crashed, to become invalid and be automatically released.

Using delayed restarts it is basically possible to achieve safety even without any kind of Redis persistence available, however note that this may translate into an availability penalty. For example if a majority of instances crash, the system will become globally unavailable for TTL (here globally means that no resource at all will be lockable during this time).

Making the algorithm more reliable: Extending the lock

If the work performed by clients is composed of small steps, it is possible to use smaller lock validity times by default, and extend the algorithm implementing a lock extension mechanism. Basically the client, if in the middle of the computation while the lock validity is approaching a low value, may extend the lock by sending a Lua script to all the instances that extends the TTL of the key if the key exists and its value is still the random value the client assigned when the lock was acquired.

The client should only consider the lock re-acquired if it was able to extend the lock into the majority of instances, and within the validity time (basically the algorithm to use is very similar to the one used when acquiring the lock).

However this does not technically change the algorithm, so anyway the max number of locks reacquiring attempts should be limited, otherwise one of the liveness properties is violated.

Want to help?

If you are into distributed systems, it would be great to have your opinion / analysis. Also reference implementations in other languages could be great.

Thanks in advance!

From: http://redis.io/topics/distlock

HDFS Checkpointing

Understanding how checkpointing works in HDFS can make the difference between a healthy cluster or a failing one.

Checkpointing is an essential part of maintaining and persisting filesystem metadata in HDFS. It’s crucial for efficient NameNode recovery and restart, and is an important indicator of overall cluster health. However, checkpointing can also be a source of confusion for operators of Apache Hadoop clusters.

In this post, I’ll explain the purpose of checkpointing in HDFS, the technical details of how checkpointing works in different cluster configurations, and then finish with a set of operational concerns and important bug fixes concerning this feature.

Filesystem Metadata in HDFS

To start, let’s first cover how the NameNode persists filesystem metadata.

HDFS metadata changes are persisted to the edit log.

At a high level, the NameNode’s primary responsibility is storing the HDFS namespace. This means things like the directory tree, file permissions, and the mapping of files to block IDs. It’s important that this metadata (and all changes to it) are safely persisted to stable storage for fault tolerance.

This filesystem metadata is stored in two different constructs: the fsimage and the edit log. The fsimage is a file that represents a point-in-time snapshot of the filesystem’s metadata. However, while the fsimage file format is very efficient to read, it’s unsuitable for making small incremental updates like renaming a single file. Thus, rather than writing a new fsimage every time the namespace is modified, the NameNode instead records the modifying operation in the edit log for durability. This way, if the NameNode crashes, it can restore its state by first loading the fsimage then replaying all the operations (also called edits or transactions) in the edit log to catch up to the most recent state of the namesystem. The edit log comprises a series of files, called edit log segments, that together represent all the namesystem modifications made since the creation of the fsimage.

As an aside, this pattern of using a log for incremental changes on top of another storage format is quite common for traditional filesystems. Log-structured filesystems take this to an extreme and use just a log for persisting data, but more common journaling filesystems like EXT3, EXT4, and XFS support writing changes to a journal before applying them to their final locations on disk.

Why is Checkpointing Important?

A typical edit ranges from 10s to 100s of bytes, but over time enough edits can accumulate to become unwieldy. A couple of problems can arise from these large edit logs. In extreme cases, it can fill up all the available disk capacity on a node, but more subtly, a large edit log can substantially delay NameNode startup as the NameNode reapplies all the edits. This is where checkpointing comes in.

Checkpointing is a process that takes an fsimage and edit log and compacts them into a new fsimage. This way, instead of replaying a potentially unbounded edit log, the NameNode can load the final in-memory state directly from the fsimage. This is a far more efficient operation and reduces NameNode startup time.

Checkpointing creates a new fsimage from an old fsimage and edit log.

However, creating a new fsimage is an I/O- and CPU-intensive operation, sometimes taking minutes to perform. During a checkpoint, the namesystem also needs to restrict concurrent access from other users. So, rather than pausing the active NameNode to perform a checkpoint, HDFS defers it to either the SecondaryNameNode or Standby NameNode, depending on whether NameNode high-availability is configured. The mechanics of checkpointing differs depending on if NameNode high-availability is configured; we’ll cover both.

In either case though, checkpointing is triggered by one of two conditions: if enough time has elapsed since the last checkpoint (dfs.namenode.checkpoint.period), or if enough new edit log transactions have accumulated (dfs.namenode.checkpoint.txns). The checkpointing node periodically checks if either of these conditions are met (dfs.namenode.checkpoint.check.period), and if so, kicks off the checkpointing process.

Checkpointing with a Standby NameNode

Checkpointing is actually much simpler when dealing with an HA setup, so let’s cover that first.

When NameNode high-availability is configured, the active and standby NameNodes have a shared storage where edits are stored. Typically, this shared storage is an ensemble of three or more JournalNodes, but that’s abstracted away from the checkpointing process.

The standby NameNode maintains a relatively up-to-date version of the namespace by periodically replaying the new edits written to the shared edits directory by the active NameNode. As a result, checkpointing is as simple as checking if either of the two preconditions are met, saving the namespace to a new fsimage (roughly equivalent to running hdfs dfsadmin -saveNamespace on the command line), then transferring the new fsimage to the active namenode via HTTP.

Checkpointing with NameNode HA configured

Here, Standby NameNode is abbreviated as SbNN and Active NameNode as ANN:

1. SbNN checks whether either of the two preconditions are met: elapsed time since the last checkpoint or number of accumulated edits.
2. SbNN saves its namespace to an a new fsimage with the intermediate namefsimage.ckpt_, where txid is the transaction ID of the most recent edit log transaction. Then, the SbNN writes an MD5 file for the fsimage, and renames the fsimage to fsimage_. While this is taking place, most other SbNN operations are blocked. This means administrative operations like NameNode failover or accessing parts of the SbNN’s webui. Routine HDFS client operations (such as listing, reading, and writing files) are unaffected as these operations are serviced by the ANN.
3. SbNN sends an HTTP GET to the active NN’s GetImageServlet at /getimage?putimage=1. The URL parameters also have the transaction ID of the new fsimage and the SbNN’s hostname and HTTP port.
4. The active NN’s servlet uses the information in the GET request to in turn do its own GET back to the SbNN’s GetImageServlet. Similar to the standby, it first saves the new fsimage with the intermediate name fsimage.ckpt_, creates the MD5 file for the fsimage, and then renames the new fsimage to fsimage_.

Checkpointing with a SecondaryNameNode

In a non-HA deployment, checkpointing is done on the SecondaryNameNode rather than the standby NameNode. Since there isn’t a shared edits directory or automatic tailing of the edit log, the SecondaryNameNode has to go through a few more steps first to refresh its view of the namespace before continuing down the same basic steps.

Time diagram of 2NN and NN with annotated steps

Here, the NameNode is abbreviated as NN and the SecondaryNameNode as 2NN:

1. 2NN checks whether either of the two preconditions are met: elapsed time since the last checkpoint or number of accumulated edits.
• In the absence of a shared edit directory, the most recent edit log transaction ID needs to be queried via an explicit RPC to the NameNode (NamenodeProtocol#getTransactionId).
2. 2NN triggers an edit log roll, which ends the current edit log segment and starts a new one. The NN can keep writing edits to the new segment while the SNN compacts all the previous ones. This also returns the transaction IDs of the current fsimage and the edit log segment that was just rolled. Explicit triggering of an edit log roll is not necessary in an HA configuration, since the standby NameNode periodically rolls the edit log orthogonal to checkpointing.
3. Given these two transaction IDs, the 2NN fetches new fsimage and edit files as needed via GET to the NN’s GetImageServlet. The 2NN might already have some of these files from a previous checkpoint (such as the current fsimage).
4. If necessary, the 2NN reloads its namespace from a newly downloaded fsimage.
5. The 2NN replays the new edit log segments to catch up to the current transaction ID.
   From here, the rest is the same as in the HA case with a StandbyNameNode.
6. 2NN writes out its namespace to a new fsimage.
7. The 2NN contacts the NN via HTTP GET at /getimage?putimage=1, causing the NN’s servlet to do its own GET to the 2NN to download the new fsimage.

Operational Implications

The biggest operational concern related to checkpointing is when it fails to happen. We’ve seen scenarios where NameNodes accumulated hundreds of GBs of edit logs, and no one noticed until the disks filled completely and crashed the NN. When this happens, there’s not much to do besides restart the NN and wait for it to replay all the edits. Because of the potential severity of this issue, Cloudera Manager will warn if the current fsimage is out of date, if the checkpointing 2NN or SbNN is down, as well as if NN disks are close to capacity.

Checkpointing is a very I/O and network intensive operation and can affect client performance. This is especially true on a large cluster with millions of files and a multi-GB fsimage, since copying a new fsimage to the NameNode can eat up all available bandwidth. In this case, the transfer speed can be throttled withdfs.image.transfer.bandwidthPerSec. If you do adjust this parameter, you might also need to adjust dfs.image.transfer.timeout based on your expected transfer time.

Conclusion

Checkpointing is a vital part of healthy HDFS operation. In this post, you learned how filesystem metadata is persisted in HDFS, the importance of checkpointing to this role, how checkpointing works in both HA and non-HA setups, and finally covered a selection of important fixes and improvements related to checkpointing.

By understanding the purpose of checkpointing and how it works, you are now equipped with the knowledge to debug these kinds of issues in your production clusters.

From: http://blog.cloudera.com/blog/2014/03/a-guide-to-checkpointing-in-hadoop/