Key-Value Store
A key-value store, also referred to as a key-value database, is a non-relational database. Each unique identifier is stored as a key with its associated value. This data pairing is knwon as a “key-value” pair.
The value in a key-value pair can be strings, lists, objects, etc. The value is usually treated as an opaque object in key-value stores.
Distributed key-value store
A distributed key-value store is also called a Distributed hash table, which distributes key-value pairs across many servers. When designing a distributed system, it is important to understand CAP theorem.
CAP theorem
CAP theorem states it is impossible for a distributed system to simultaneously provide more than two of these three guarantees: consistency, availability, and partition tolerance as follows:
CP (consistency and partition tolerance) systems
AP (availability and partition tolerance) systems
CA (consistency and availability tolerance) systems
Since network failure is unavoidable, a distributed system must tolerate network partition. Thus, a CA system cannot exist in real-world applications.
Data Partition
05-Design Consistent Hashing > Consistent Hashing
Data Replication
To achieve high availability and reliability, data must be replicated asynchronously over M servers, where m is a configurable perameter.
With virtual nodes, the first M servers on the ring may be owned by fewer than M phyiscal servers. To avoid this issue, we only choose unique servers while performing the clockwise walk logic. For better reliability, replicas are placed in distant data centers, and data centers are connected through high-speed networks.
05-Distributed K-V Store & Amazon Dynamo > Preference list (data replication)
Consistency
Since data is replicateed at multiple nodes, it must be synchronized across replicas. Quorum consensus can guarantee consistency for both read and write operations.
Consistency Model
Consistency model is other important factor to consider when designing a key-value store. A consistency model defines the degree of data consistency, and a wide spectrum os possible consistency model exists.
Strong consistency is usually achieved by forcing a replica not to accept new reads/writes until every replica has agreed on current write.
Inconsistency resolution: versioning
Replication gives high availability but causes inconsistenceis among replicas. Versioning and vector clocks are used to solve inconsistency.
05-Distributed K-V Store & Amazon Dynamo > An example of conflicting writes (versions)
When a client reads D3 and D4, it discovers a conflict, which is caused by data item D2 being modified by both Sy and Sz. The conflict is resolved by the client and updated data is sent to the server. Assume the write is handled by Sx, which now has D5([Sx, 3], [Sy, 1], [Sz, 1]).
Even though vector clocks can resolve conflicts, there are two notable downsides.
First, vector clocks add complexity to the client because it needs to implement conflict resolution logic.
Second, the [server: version] pairs in the vector clock could grow rapidly. To fix this problem, we set a threshold for the length, and if it exceeds the limit, the oldest pairs are removed. This can lead to inefficiencies in reconciliation because the descendant relationship cannot be determined accurately. However, based on Dynamo paper, Amazon has not yet encountered this problem in production; therefore, it is probably an acceptable solution for most companies.
Handling Failures
Failure Detection
In a distributed system, it is insufficient to believe that a server is down because another server says so. Usually, it requires at least two independent sources of information to mark a server down.
All-to-all multicasting is a straightforward solution. However, this is inefficient when many servers are in the system.
A better solution is to use decentralized failure detection methods like gossip protocol. Gossip protocl works as follows:
- Each node maintains a node membership list, which contains member IDs and heartbeat counters.
- Each node periodically increments its heartbeat counter.
- Each node periodically sends heartbeats to a set of random nodes, which in turn propagate to another set of nodes.
- Once nodes receive heartbeats, membership list is updated to the latest info.
- If the heartbeat has not increased for more than predefined periods, the member is considered as offline.
Handling temporary failure
05-Distributed K-V Store & Amazon Dynamo > Failure Handling sloppy quorum + hinted handoff
Handling permanent failure
What if a replica is permanently unavailable? To handle such a situation, we implement an anti-entropy protocol to keep replicas in sync. Anti-entropy involves comparing each piece of data on replicas and updating each replica to the newest version. A Merkle tree is used for inconsistency detection and minimizing the amount of data transferred.
Assuming key space is from 1 to 12, the following steps show how to build a Merkle tree. Highlighted boxes indicate inconsistency.
Step 1: Divide key space into buckets (4 in our example) as shown below. A bucket is used as the root level node to maintain a limited depth of the tree:
Step 2: Once the buckets are created, hash each key in a bucket using a uniform hashing method:
Step 3: Create a single hash node per bucket:
Step 4: Build the tree upwards till root by calculating hashes of children:
To compare two Merkle trees, start by comparing the root hashes. If root hashes match, both servers have the same data. If root hashes disagree, then the left child hashes are compared followed by right child hashes. You can traverse the tree to find which buckets are not synchronized and synchronize those buckets only. Using Merkle trees, the amount of data needed to be synchronized is proportional to the differences between the two replicas, and not the amount of data they contain. In real-world systems, the bucket size is quite big. For instance, a possible configuration is one million buckets per one billion keys, so each bucket only contains 1000 keys.