Tag: 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 | 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.
DDIA | Replication | Multi Leader Replication | Handling write conflicts

October 7th, 2023
In previous post, we learnt about multi-leader replication and the use cases where we need that multi-leader replication. In this post we will look into a common problem with multi leader replication – write conflicts.

Consider google doc which is shared between multiple people, if two people edit the same line at the same time from two different machines, in multi leader replication, both of them will look like they are written on the replicas, but there is a conflict which needs to be solved before committing them. At this time though, it will be too late to resolve the conflict by asking the users.

We could make conflict detection synchronous, wait for the write to be replicated to all replicas before telling the user that the write was successful.

However, by doing so, you would lose the main advantage of multi-leader replication: allowing each replica to accept writes independently. If you want synchronous conflict detection, you might as well just use single-leader replication.

Conflict avoidance

The simplest strategy for dealing with conflicts is to avoid them: if the application can ensure that all writes for a particular record go through the same leader, then conflicts cannot occur.

Say we have two people A and B. A’s writes will always go to the leader A (one data center) and B’s writes would always go to the leader B (second datacenter). Same for reads, A will read from leader A and B will read from leader B. Different users may have different “home” data centers (perhaps picked based on geographic proximity to the user), but from any one user’s point of view the configuration is essentially single-leader.

However conflict avoidance breaks when either data center fails/user moves to another data center’s region and now we need to change the leader to another data center, that leader might not have the whole data from previous leaders and there is possibility of concurrent writes as both users now have same leader.

Converging toward a consistent state

A single-leader database applies writes in a sequential order: if there are several updates to the same field, the last write determines the final value of the field.

In a multi-leader configuration, there is no defined ordering of writes, so it’s not clear what the final value should be. The database must resolve the conflict in a convergent way, which means that all replicas must arrive at the same final value when all changes have been replicated.

There are various ways of achieving convergent conflict resolution:
- Give each write a unique ID, pick the write with the highest ID as the winner, and throw away the other writes. If a timestamp is used, this technique is known as last write wins (LWW). Although this approach is popular, it is dangerously prone to data loss. We will discuss LWW in more detail at the end of this chapter.
- Give each replica a unique ID, and let writes that originated at a higher-numbered replica always take precedence over writes that originated at a lower-numbered replica. This approach also implies data loss.
- Somehow merge the values together—e.g., order them alphabetically and then concatenate them.
- Record the conflict in an explicit data structure that preserves all information, and write application code that resolves the conflict at some later time (perhaps by prompting the user).
Custom conflict resolution logic

Letting application handle conflict resolution either at write or at read.

On write

As soon as the database system detects a conflict in the log of replicated changes, it calls the conflict handler.

On read

When a conflict is detected, all the conflicting writes are stored. The next time the data is read, these multiple versions of the data are returned to the application. The application may prompt the user or automatically resolve the conflict, and write the result back to the database. CouchDB works this way, for example.

Automatic conflict resolution

Different ways in which we can achieve automatic conflict resolution in multi leader database –
- Conflict-free replicated datatypes (CRDTs) are a family of data structures for sets, maps (dictionaries), ordered lists, counters, etc. that can be concurrently edited by multiple users, and which automatically resolve conflicts in sensible ways. Some CRDTs have been implemented in Riak 2.0.
- Mergeable persistent data structures track history explicitly, similarly to the Git version control system, and use a three-way merge function (whereas CRDTs use two-way merges).
- Operational transformation is the conflict resolution algorithm behind collaborative editing applications such as Etherpad and Google Docs. It was designed particularly for concurrent editing of an ordered list of items, such as the list of characters that constitute a text document.
Thanks for stopping by! Hope this gives you a brief overview in to handling conflicts for multi-leader replication. Eager to hear your thoughts and chat, please leave comments below and we can discuss.
DDIA | Replication | Multi Leader Replication

September 30th, 2023
In previous posts, we learned about single leader based replication, different types and problems associated with this approach. In single leader based replication, there is only one leader who accepts writes. There are systems where there are multiple leaders/multiple nodes that accept writes. Those systems are using multi-Leader replication. Let’s see some usecases for multi-leader replication.

Multi-datacenter operation

There are multiple data centers for a given system. If there is only one leader then writes from other data centers would need to come to the leader which causes addition to latency as both data centers are far apart.

To solve this, we can have one leader per data center. Within each datacenter, regular leader–follower replication is used; between data centers, each datacenter’s leader replicates its changes to the leaders in other data centers.

Advantages –

1. Better performance, Lower latency, as data is replicated within a data center.

2. High tolerance, if a leader in one data center goes out, other data centers are not affected.

3. High tolerance to network related failures

Disadvantages –
1. the same data may be concurrently modified in two different data centers, and those write conflicts must be resolved
Some databases support multi-leader configurations by default, but it is also often implemented with external tools, such as Tungsten Replicator for MySQL, BDR for PostgreSQL, and GoldenGate for Oracle.

Clients with offline operation

Another situation in which multi-leader replication is appropriate is if you have an application that needs to continue to work while it is disconnected from the internet.

For example, consider the calendar apps on your mobile phone, your laptop, and other devices. You need to be able to see your meetings (make read requests) and enter new meetings (make write requests) at any time, regardless of whether your device currently has an internet connection. If you make any changes while you are offline, they need to be synced with a server and your other devices when the device is next online.

In this case, every device has a local database that acts as a leader (it accepts write requests), and there is an asynchronous multi-leader replication process (sync) between the replicas of your calendar on all of your devices. The replication lag may be hours or even days, depending on when you have internet access available.

Collaborative editing

Real-time collaborative editing applications allow several people to edit a document simultaneously. When one user edits a document, the changes are instantly applied to their local replica (the state of the document in their web browser or client application) and asynchronously replicated to the server and any other users who are editing the same document.

If you want to guarantee that there will be no editing conflicts, the application must obtain a lock on the document before a user can edit it. If another user wants to edit the same document, they first have to wait until the first user has committed their changes and released the lock. This collaboration model is equivalent to single-leader replication with transactions on the leader.

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

Tag: 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 | Multi Leader Replication | Multi-Leader Replication Topologies

DDIA | Replication | Multi Leader Replication | Handling write conflicts

Conflict avoidance

Converging toward a consistent state

Custom conflict resolution logic

Automatic conflict resolution

DDIA | Replication | Multi Leader Replication

Multi-datacenter operation

Clients with offline operation

Collaborative editing