Category: DDIA

DDIA | Partitioning | Rebalancing Partitions

November 18th, 2023
In previous posts, we learned about partitioning schemes for key-value pairs and secondary indexes, in all of these schemes there is a need for redistributing data either when there is a new node or a new partition. In this post, we are going to touch upon the techniques to do handle that data migration.

The process of moving load from one node in the cluster to another is called rebalancing..

Some requirements for rebalancing –
- After rebalancing, the load (data storage, read and write requests) should be shared fairly between the nodes in the cluster.
- While rebalancing is happening, the database should continue accepting reads and writes.
- No more data than necessary should be moved between nodes, to make rebalancing fast and to minimize the network and disk I/O load.
Strategies for Rebalancing
1. Fixed number of partitions – create many more partitions than there are nodes, and assign several partitions to each node. Only entire partitions are moved between nodes. if a node is added to the cluster, the new node can steal a few partitions from every existing node until partitions are fairly distributed once again. The number of partitions does not change, nor does the assignment of keys to partitions. The only thing that changes is the assignment of partitions to nodes. This change of assignment is not immediate—it takes some time to transfer a large amount of data over the network—so the old assignment of partitions is used for any reads and writes that happen while the transfer is in progress.
2. Dynamic partitioning – Key range–partitioned databases such as HBase and RethinkDB create partitions dynamically. When a partition grows to exceed a configured size (on HBase, the default is 10 GB), it is split into two partitions so that approximately half of the data ends up on each side of the split. Conversely, if lots of data is deleted and a partition shrinks below some threshold, it can be merged with an adjacent partition. This process is similar to what happens at the top level of a B-tree.
  
  Each partition is assigned to one node, and each node can handle multiple partitions, like in the case of a fixed number of partitions. After a large partition has been split, one of its two halves can be transferred to another node in order to balance the load. In the case of HBase, the transfer of partition files happens through HDFS, the underlying distributed filesystem.
  
  An advantage of dynamic partitioning is that the number of partitions adapts to the total data volume. If there is only a small amount of data, a small number of partitions is sufficient, so overheads are small; if there is a huge amount of data, the size of each individual partition is limited to a configurable maximum.
3. Partitioning proportionally to nodes – make the number of partitions proportional to the number of nodes—in other words, to have a fixed number of partitions per node. In this case, the size of each partition grows proportionally to the dataset size while the number of nodes remains unchanged, but when you increase the number of nodes, the partitions become smaller again. Since a larger data volume generally requires a larger number of nodes to store, this approach also keeps the size of each partition fairly stable.
Operations: Automatic or Manual Rebalancing

Automatic rebalancing – the system decides automatically when to move partitions from one node to another, without any administrator interaction. Fully automated rebalancing can be convenient, because there is less operational work to do for normal maintenance. However, it can be unpredictable.

Rebalancing is an expensive operation, because it requires rerouting requests and moving a large amount of data from one node to another. If it is not done carefully, this process can overload the network or the nodes and harm the performance of other requests while the rebalancing is in progress. SO it’s better to have human in the loop ie use the manual rebalancing.

Manual rebalancing – the assignment of partitions to nodes is explicitly configured by an administrator, and only changes when the administrator explicitly reconfigures it.

Thanks for stopping by! Hope this gives you a brief overview in to rebalancing partitions. Eager to hear your thoughts and chat, please leave comments below and we can discuss.
DDIA | Partitioning | Partitioning and Secondary Indexes

November 13th, 2023

In previous post we learned about the partitioning schemes based on key-value pairs.

If records are only ever accessed via their primary key, we can determine the partition from that key and use it to route read and write requests to the partition responsible for that key.

The situation becomes more complicated if secondary indexes are involved as they are not unique. The problem with secondary indexes is that they don’t map neatly to partitions. There are two main approaches to partitioning a database with secondary indexes: document-based partitioning and term-based partitioning.

Partitioning Secondary Indexes by Document

Let’s take the example given in the book. For example, we are running a website to buy/sell used cars. Each listing has unique ID called DocumentID and we partition the database by documentID.

If we want our users to search by make or color, then we will need to add them as secondary indexes on the database. Database takes care of indexing if we have declared the index. It will assign the new entries to the correct partition.

Whenever you write to the database—to add, remove, or update a document—you only need to deal with the partition that contains the document ID that you are writing. For that reason, a document-partitioned index is also known as a local index .

However, reading from a document-partitioned index requires care: f you want to search using secondary index, you need to send the query to all partitions, and combine all the results you get back, also known as scatter/gather. it can make read queries on secondary indexes quite expensive. Even if you query the partitions in parallel, scatter/gather is prone to tail latency amplification.

Nevertheless, it is widely used: MongoDB, Riak, Cassandra, Elasticsearch, SolrCloud, and VoltDB all use document-partitioned secondary indexes.

Partitioning Secondary Indexes by Term

Rather than each partition having its own secondary index (a local index), we can construct a global index that covers data in all partitions. However, we can’t just store that index on one node, since it would likely become a bottleneck and defeat the purpose of partitioning. A global index must also be partitioned, but it can be partitioned differently from the primary key index.

For example, red cars from all partitions appear under color:red in the index, but the index is partitioned so that colors starting with the letters a to r appear in partition 0 and colors starting with s to z appear in partition 1. The index on the make of car is partitioned similarly (with the partition boundary being between f and h).

As before, we can partition the index by the term itself, or using a hash of the term. Partitioning by the term itself can be useful for range scans (e.g., on a numeric property, such as the asking price of the car), whereas partitioning on a hash of the term gives a more even distribution of load.

The advantage of a global (term-partitioned) index over a document-partitioned index is that it can make reads more efficient: rather than doing scatter/gather over all partitions, a client only needs to make a request to the partition containing the term that it wants.

However, the downside of a global index is that writes are slower and more complicated, because a write to a single document may now affect multiple partitions of the index (every term in the document might be on a different partition, on a different node).

Thanks for stopping by! Hope this gives you a brief overview in to partitioning and secondary indexes. Eager to hear your thoughts and chat, please leave comments below and we can discuss.
DDIA | Partitioning | Partitioning of Key-Value Data

November 4th, 2023
In previous few posts, we learned about Replication, from this post, we will look into partitions.

Replication is having multiple copies of same data on multiple nodes. For very large datasets, or very high query throughput, that is not sufficient: we need to break the data up into partitions, also known as sharding.

The main reason for wanting to partition data is scalability. Different partitions can be placed on different nodes in a shared-nothing cluster. Thus, a large dataset can be distributed across many disks, and the query load can be distributed across many processors.

Partitioning of Key-Value Data

Skewed partitioning – If the partitioning is unfair, so that some partitions have more data or queries than others, we call it skewed

Hotspot – all the load could end up on one partition, so 9 out of 10 nodes are idle and your bottleneck is the single busy node. A partition with disproportionately high load is called a hot spot.
1. Partitioning by Key Range – One way of partitioning is to assign a continuous range of keys (from some minimum to some maximum) to each partition. If you know the boundaries between the ranges, you can easily determine which partition contains a given key. If you also know which partition is assigned to which node, then you can make your request directly to the appropriate node.
  
  The downside of key range partitioning is that certain access patterns can lead to hot spots. If the key is a timestamp, then the partitions correspond to ranges of time, and it’s possible for some of the data to be accessed at particular times while nothing gets accessed in other times.
2. Partitioning by Hash of Key – Using hash function to determine partition of keys. Once you have a suitable hash function for keys, you can assign each partition a range of hashes (rather than a range of keys), and every key whose hash falls within a partition’s range will be stored in that partition.
  
  Unfortunately however, by using the hash of the key for partitioning we lose a nice property of key-range partitioning: the ability to do efficient range queries. Keys that were once adjacent are now scattered across all the partitions, so their sort order is lost. This can be solved using compound primary key consisting of several columns. Only the first part of that key is hashed to determine the partition, but the other columns are used as a concatenated index for sorting the data. (used by cassandra).
Skewed Workloads and Relieving Hot Spots

Consider this example – for example, on a social media site, a celebrity user with millions of followers may cause a storm of activity when they do something. This event can result in a large volume of writes to the same key (where the key is perhaps the user ID of the celebrity, or the ID of the action that people are commenting on). Hashing the key doesn’t help, as the hash of two identical IDs is still the same.

Today, most data systems are not able to automatically compensate for such a highly skewed workload, so it’s the responsibility of the application to reduce the skew.

For example, if one key is known to be very hot, a simple technique is to add a random number to the beginning or end of the key. However, having split the writes across different keys, any reads now have to do additional work, as they have to read the data from all keys and combine it.

This technique also requires additional bookkeeping: it only makes sense to append the random number for the small number of hot keys; for the vast majority of keys with low write throughput this would be unnecessary overhead. Thus, you also need some way of keeping track of which keys are being split.

Thanks for stopping by! Hope this gives you a brief overview in to partitioning. Eager to hear your thoughts and chat, please leave comments below and we can discuss.
DDIA | Replication | Leaderless Replication | Sloppy Quorums | Hinted Handoff | Multi-datacenter operations

October 28th, 2023
In previous posts, we looked into leaderless replications and quorums. In this post we continue learning about them.

Quorums are not as fault-tolerant as they could be. A network interruption can easily cut off a client from a large number of database nodes.

Although those nodes are alive, and other clients may be able to connect to them, to a client that is cut off from the database nodes, they might as well be dead. In this situation, it’s likely that fewer than w or r reachable nodes remain, so the client can no longer reach a quorum.

In a large cluster (with significantly more than n nodes) it’s likely that the client can connect to some database nodes during the network interruption, just not to the nodes that it needs to assemble a quorum for a particular value. In that case, database designers face a trade-off:
- Is it better to return errors to all requests for which we cannot reach a quorum of w or r nodes?
- Or should we accept writes anyway, and write them to some nodes that are reachable but aren’t among the n nodes on which the value usually lives?
Sloppy Quorum – writes and reads still require w and r successful responses, but those may include nodes that are not among the designated n “home” nodes for a value.

Hinted Handoff – Once the network interruption is fixed, any writes that one node temporarily accepted on behalf of another node are sent to the appropriate “home” nodes. This is called hinted handoff.

Sloppy Quorums only assures that writes are successful given there are w nodes available somewhere. Reads could still be returning stale value until hinted handoff completes.

Sloppy quorums are optional in dynamo implementations, for Riak they are turned on.

Multi-datacenter operation

Leaderless replication is also suitable for multi-datacenter operation, since it is designed to tolerate conflicting concurrent writes, network interruptions, and latency spikes.

Cassandra and Voldemort – the number of replicas n includes nodes in all datacenters, and in the configuration you can specify how many of the n replicas you want to have in each datacenter. Each write from a client is sent to all replicas, regardless of datacenter, but the client usually only waits for acknowledgment from a quorum of nodes within its local datacenter so that it is unaffected by delays and interruptions on the cross-datacenter link. The higher-latency writes to other datacenters are often configured to happen asynchronously, although there is some flexibility in the configuration

Detecting Concurrent Writes

Dynamo-style databases allow several clients to concurrently write to the same key, which means that conflicts will occur even if strict quorums are used. Conflicts also occur due to read repair and hinted handoffs.

Application developers have to make sure these conflicts are resolved as databases don’t do that automatically.
1. Last write wins (discarding concurrent writes) – One approach for achieving eventual convergence is to declare that each replica need only store the most “recent” value and allow “older” values to be overwritten and discarded. For example, we can attach a timestamp to each write, pick the biggest timestamp as the most “recent,” and discard any writes with an earlier timestamp. This conflict resolution algorithm, called last write wins (LWW). LWW achieves the goal of eventual convergence, but at the cost of durability: if there are several concurrent writes to the same key, even if they were all reported as successful to the client (because they were written to w replicas), only one of the writes will survive and the others will be silently discarded.
2. The “happens-before” relationship and concurrency – whenever you have two operations A and B, there are three possibilities: either A happened before B, or B happened before A, or A and B are concurrent. What we need is an algorithm to tell us whether two operations are concurrent or not. If one operation happened before another, the later operation should overwrite the earlier operation, but if the operations are concurrent, we have a conflict that needs to be resolved.
  
  The algorithm works as follows:
  - The server maintains a version number for every key, increments the version number every time that key is written, and stores the new version number along with the value written.
  - When a client reads a key, the server returns all values that have not been overwritten, as well as the latest version number. A client must read a key before writing.
  - When a client writes a key, it must include the version number from the prior read, and it must merge together all values that it received in the prior read. (The response from a write request can be like a read, returning all current values, which allows us to chain several writes like in the shopping cart example.)
  - When the server receives a write with a particular version number, it can overwrite all values with that version number or below (since it knows that they have been merged into the new value), but it must keep all values with a higher version number (because those values are concurrent with the incoming write).
  - If you make a write without including a version number, it is concurrent with all other writes, so it will not overwrite anything—it will just be returned as one of the values on subsequent reads.
  - If several operations happen concurrently, clients have to clean up afterward by merging the concurrently written values, also called sibling values. Merging sibling values is essentially the same problem as conflict resolution in multi-leader replication. A simple approach is to just pick one of the values based on a version number or timestamp (last write wins), but that implies losing data. Another option is to have a union of the two values, but that is also error prone. We will discuss later approaches for handling these sibling values.
3. Version Vectors – When there are multiple replicas, we need to use a version number per replica as well as per key. Each replica increments its own version number when processing a write, and also keeps track of the version numbers it has seen from each of the other replicas. This information indicates which values to overwrite and which values to keep as siblings. The collection of version numbers from all the replicas is called a version vector.
Thanks for stopping by! Hope this gives you a brief overview in to sloppy quorums, hinted hand-offs, multi-data center operations and detecting concurrent writes when we are using leaderless replication. Eager to hear your thoughts and chat, please leave comments below and we can discuss.
DDIA | Replication | Leaderless Replication

October 21st, 2023
So far in previous posts we have read about leader based replication. There is another way of handling replication and that is leaderless. It didn’t become popular until Amazon used it for DynamoDb.

As the name suggest there is no leader which accepts writes, instead client can send write requests to multiple replicas. In some leaderless implementations, the client directly sends its writes to several replicas, while in others, a coordinator node does this on behalf of the client.

Not only writes, but the read requests are also send to all replicas in parallel. The client may get different responses from different nodes; i.e., the up-to-date value from one node and a stale value from another. Version numbers are used to determine which value is newer .

Let’s say we have three nodes, and one node went down. Meanwhile we get a write request, this requests gets written to the two healthy replicas. Now the down replica comes back. How do we ensure that it gets the latest writes from other nodes?

Two mechanisms are often used in Dynamo-style datastores:
1. Read repair – When a client makes a read from several nodes in parallel, it can detect any stale responses. And then can write the new data to the node from which it gets stale response. This approach works well for values that are frequently read.
2. Anti-entropy process – In addition, some datastores have a background process that constantly looks for differences in the data between replicas and copies any missing data from one replica to another. Unlike the replication log in leader-based replication, this anti-entropy process does not copy writes in any particular order, and there may be a significant delay before data is copied.
Quorums for reading and writing

If there are n replicas, every write must be confirmed by w nodes to be considered successful, and we must query at least r nodes for each read. As long as w + r > n, we expect to get an up-to-date value when reading, because at least one of the r nodes we’re reading from must be up to date.

In Dynamo-style databases, the parameters n, w, and r are typically configurable. A common choice is to make n an odd number (typically 3 or 5) and to set w = r = (n + 1) / 2 (rounded up).

The quorum condition, w + r > n, allows the system to tolerate unavailable nodes as follows:
- - If w < n, we can still process writes if a node is unavailable.
  - If r < n, we can still process reads if a node is unavailable.
  - With n = 3, w = 2, r = 2 we can tolerate one unavailable node.
  - With n = 5, w = 3, r = 3 we can tolerate two unavailable nodes.
  - Normally, reads and writes are always sent to all n replicas in parallel. The parameters w and r determine how many nodes we wait for—i.e., how many of the n nodes need to report success before we consider the read or write to be successful.
Limitations –

With a smaller w and r you are more likely to read stale values, because it’s more likely that your read didn’t include the node with the latest value.

However, even with w + r > n, there are likely to be edge cases where stale values are returned. These depend on the implementation, but possible scenarios include:
- If a sloppy quorum is used, the w writes may end up on different nodes than the r reads, so there is no longer a guaranteed overlap between the r nodes and the w nodes.
- If two writes occur concurrently, it is not clear which one happened first. In this case, the only safe solution is to merge the concurrent writes. If a winner is picked based on a timestamp (last write wins), writes can be lost due to clock skew.
- If a write happens concurrently with a read, the write may be reflected on only some of the replicas. In this case, it’s undetermined whether the read returns the old or the new value.
- If a write succeeded on some replicas but failed on others, and overall succeeded on fewer than w replicas, it is not rolled back on the replicas where it succeeded. This means that if a write was reported as failed, subsequent reads may or may not return the value from that write.
- If a node carrying a new value fails, and its data is restored from a replica carrying an old value, the number of replicas storing the new value may fall below w, breaking the quorum condition.
- Even if everything is working correctly, there are edge cases in which you can get unlucky with the timing.
Thanks for stopping by! Hope this gives you a brief overview in to leaderless replication. Eager to hear your thoughts and chat, please leave comments below and we can discuss.
DDIA | Replication | Multi Leader Replication | Multi-Leader Replication Topologies

October 14th, 2023

In previous post, we ;looked into how to handle write conflicts in multi leader replication. This post concludes the multi leader replication with an insight into topologies.

A replication topology describes the communication paths along which writes are propagated from one node to another. If you have two leaders, there is only one plausible topology: leader 1 must send all of its writes to leader 2, and vice versa. With more than two leaders, various different topologies are possible.

All-to-all – every leader sends its writes to every other leader.

Circular – each node receives writes from one node and forwards those writes (plus any writes of its own) to one other node.

Star – one designated root node forwards writes to all of the other nodes.

In circular and star topologies, a write may need to pass through several nodes before it reaches all replicas. Therefore, nodes need to forward data changes they receive from other nodes.

To prevent infinite replication loops, each node is given a unique identifier, and in the replication log, each write is tagged with the identifiers of all the nodes it has passed through. So when node receives data tagged with it’s own identifier it igones that data and breaks the cycle.

A problem with circular and star topologies is that if just one node fails, it can interrupt the flow of replication messages between other nodes, causing them to be unable to communicate until the node is fixed.

A problem with all to all topology is that sequence of writes could be broken resulting in faulty data if conflict resolution strategies is not implemented.

Thanks for stopping by! Hope this gives you a brief overview in to different multi leader replication topologies and their problems.. Eager to hear your thoughts and chat, please leave comments below and we can discuss.

Category: DDIA

DDIA | Partitioning | Rebalancing Partitions

Strategies for Rebalancing

Operations: Automatic or Manual Rebalancing

DDIA | Partitioning | Partitioning and Secondary Indexes

Partitioning Secondary Indexes by Document

Partitioning Secondary Indexes by Term

DDIA | Partitioning | Partitioning of Key-Value Data

Partitioning of Key-Value Data

Skewed Workloads and Relieving Hot Spots

DDIA | Replication | Leaderless Replication | Sloppy Quorums | Hinted Handoff | Multi-datacenter operations

Multi-datacenter operation

Detecting Concurrent Writes

DDIA | Replication | Leaderless Replication

Quorums for reading and writing

DDIA | Replication | Multi Leader Replication | Multi-Leader Replication Topologies