Loading...
The simplest straightforward solution is to store key-value pairs in a hash table, which is how most systems work.
However, when using a hash table, we generally store everything in memory which is not always practical when the data set is large.
There are two options:
Given that we want to build a highly scalable system that could store potentially billions of records while be durable, just these two approaches are not enough. See Database section on further exploration.
getValue(key)
addKeyValue(key, value)
updateKeyValue(key, value)
deleteKeyValue(key, value)
Given the non-functional requirement of durability and scalability of a distributed key-value store we need to consider fault-tolerance and how to divide the data for scalability.
Let's talk about sharding. Assume we are using URL as keys for the sake of illustration.
One approach to sharding is to divide URLs across 26 machines based on the first character after "www". However, this method is ineffective because storage and traffic would not be evenly distributed—many more URLs start with "a" than "z", and popular sites like Facebook and Google would skew traffic further.
A better approach is to hash the URL, which distributes keys more evenly and efficiently across machines.
Data must be partitioned across a cluster of servers to prevent any single server from holding the entire dataset. Even if the data fits on one server, partitioning improves performance by ensuring that frequently accessed "hot" data can fit into memory, reducing access times.
In partitioned systems, each server holds specific data, and requests must be routed to the appropriate server. To avoid data loss when servers fail, become overloaded, or go offline, data is typically replicated across multiple servers.
Instead of using a simple K mod S approach, which requires significant data redistribution when nodes are added or removed, consistent hashing is a more efficient solution. It distributes data evenly across servers and minimizes the number of keys that need to be moved when the cluster changes. With consistent hashing, each key is mapped to a point on a circular hash ring, and each server is assigned multiple positions on this ring, ensuring load is evenly balanced. When a server is added or removed, only a small fraction of data needs to be reallocated to maintain balance.
This method allows for efficient, scalable partitioning, enabling peer-to-peer lookups without relying on a central metadata server, while minimizing the impact of cluster changes on data distribution.
In distributed systems, durability/availability is crucial, especially when servers fail due to hardware issues or software bugs. If a server fails, its data becomes inaccessible, impacting the overall system. To prevent downtime and ensure availability, replication is key. By maintaining copies of data across multiple servers, system availability improves—if one server crashes, another can serve the requests.
For example, if a server has a 10% chance of crashing monthly, having a backup reduces the risk of total failure to 1%. Replication helps prevent downtime, but unlike sharding, it doesn’t address storage limitations, as it focuses on availability rather than dividing data across servers.
Sharding distributes data across multiple machines to handle large datasets. Replication, on the other hand, ensures data availability by duplicating data across servers. Sharding solves storage limitations, while replication reduces downtime. Both approaches often work together to balance scalability and fault tolerance.
A distributed key-value store uses key techniques like sharding, replication, and consistent hashing to ensure scalability, fault tolerance, and high performance.
Sharding splits data across multiple servers, evenly distributing the load to handle larger volumes. Replication ensures reliability by copying data across nodes, allowing for seamless recovery if a server fails.
Consistent hashing efficiently distributes keys and minimizes data movement when nodes are added or removed, reducing disruptions. With in-memory storage for faster lookups, the system offers low-latency access, making it ideal for real-time applications.
Combined, these techniques provide high availability, scalability, and resilience, even in high-traffic environments.
Retrieval of value via key
Adding key value pair
We will be omitting other request flows given they are similar.
Introducing replicas improves availability but raises consistency issues. For example, if machine A1 has a replica A2, how can we ensure both have identical data? When updating an entry, both A1 and A2 must be updated. However, if one fails, A1 and A2 could diverge over time, leading to inconsistencies.
Several solutions address this problem. One approach is for the coordinator to keep a local copy of the new data. If an update fails, the coordinator can retry the operation. Another option is to use a commit log, similar to Git, where each node records every change. Before updating an entry, the change is written to the commit log, and a separate process ensures all updates are applied in order. If an operation fails, the system can replay the logs to restore consistency.
For read operations, If the desired data is found on machines A1, A2 and A3, the coordinator can request it from all three computers. If the data is different, the system will automatically resolve the disagreement. It's important to remember that none of these strategies is mutually exclusive. Depending on the application, you may want to utilize more than one.
In distributed systems, a basic versioning method like optimistic locking (where each data piece has a counter or "clock") works well in centralized databases but struggles in distributed environments. Machines may come and go, and replication delays can cause inconsistencies. For example, two recent writes might conflict with each other without the system knowing which one to discard.
To handle this, a vector clock is used, which tracks a counter for each server involved in writing. This allows the system to detect when two versions of data conflict or when one version supersedes another. For example, a vector clock could look like this: [1:45, 2:3, 5:55]. If one version’s vector is higher in every position than another, it succeeds the older one. If they can’t be compared, the two versions coexist in conflict.
Maintaining consistency becomes a challenge when writes are distributed across multiple servers or data centers. Traditional solutions like distributed transactions (e.g., two-phase commit) are slow and prone to failure, especially when requiring coordination across many nodes or data centers. This introduces high latency and makes consistency difficult to achieve.
An alternative is to tolerate temporary inconsistency and resolve conflicts during reads. This approach, known as read-repair, means we reconcile differences when reading the data rather than trying to guarantee perfect consistency during every write. In distributed systems, applications often follow a read-modify-update pattern where data is loaded, modified, and written back. During this process, consistency issues may arise due to replication delays.
For maximum efficiency and availability, we suggest to use a combination of read-repair and versioning (e.g., vector clocks). This allows for fewer network roundtrips compared to methods like two-phase commit while maintaining consistency across replicas. Hinted handoff ensures that downed nodes eventually receive updates without requiring the system to halt during failures, offering a balance between consistency, performance, and fault tolerance.
In distributed databases, balancing consistency, latency, and availability presents a complex challenge because improving one aspect often compromises another. Consistency ensures that all nodes have the same, up-to-date view of the data. This is important for guaranteeing that clients receive the most recent and accurate data, but it requires coordination across nodes. This coordination introduces latency, as updates must propagate to all replicas before a client can receive a response. When you try to maintain strong consistency, the need for this synchronization can slow down the system, particularly in geographically distributed setups where communication across regions adds additional delay.
On the other hand, reducing latency often means compromising on consistency. If you want to ensure a fast response time, the system may allow a node to return data before all replicas are fully synchronized, resulting in a potential inconsistency where different nodes may hold slightly different versions of the same data. In such cases, users get faster responses but may be reading stale data, depending on which node they are accessing.
Availability ensures that the system continues to function and process read and write requests even when some nodes are down or unreachable, such as during network partitions. However, prioritizing availability over consistency means allowing operations to proceed without guaranteeing that all replicas have the same data. As a result, some nodes may serve outdated or conflicting information during failure events. The system remains operational, but the data may not always be correct or synchronized.
The trade-offs between these factors are at the core of distributed database design. When a system prioritizes consistency, it may sacrifice speed and availability. Conversely, prioritizing availability and low latency can lead to inconsistencies across replicas, particularly during network partitions or failures. These challenges are partially addressed by theories like the CAP theorem, which highlights the trade-off between consistency and availability during a network partition, and PACELC, which adds the consideration of latency when there is no partition. However, these models, while useful, don’t fully capture the complexity of balancing these trade-offs in real-world scenarios, particularly when factoring in geographical distribution, variable network conditions, and the need to meet different application requirements.
See Database section and Detailed Component Design section.
Geo-partitioning - As our system expands globally, maintaining low-latency access for users in different regions while ensuring data consistency becomes difficult.