09 Sharding Cluster Data Grows Should We Increase Memory or Add Instances #

I once encountered a requirement: to use Redis to store 50 million key-value pairs, with each pair being approximately 512B. In order to quickly deploy and provide services externally, we used cloud servers to run Redis instances. So, how should we choose the memory capacity of the cloud server?

I roughly calculated that these key-value pairs occupy about 25GB of memory space (50 million * 512B). Therefore, at that time, my first solution was to choose a 32GB memory cloud server to deploy Redis. Because 32GB of memory can store all the data, and there is still 7GB left to ensure the normal operation of the system. At the same time, I also used RDB for data persistence to ensure that the data can be recovered from RDB in case of a Redis instance failure.

However, during usage, I found that Redis responses would sometimes be very slow. Later, we used the INFO command to check the latest_fork_usec metric value of Redis (which represents the duration of the most recent fork), and the result showed that this value was particularly high, almost reaching the scale of seconds.

This is related to Redis’s persistence mechanism. When using RDB for persistence, Redis forks a child process to complete the operation, and the time it takes for the fork operation is positively correlated with the amount of Redis data. Furthermore, the fork operation blocks the main thread during execution. The larger the data volume, the longer the main thread is blocked by the fork operation. Therefore, when using RDB to persist 25GB of data, the large data volume causes the background running child process to block the main thread during fork creation, resulting in slower Redis responses.

It seems that the first solution is obviously not feasible, and we must find other solutions. At this point, we noticed the concept of Redis sharding cluster. Although building a sharding cluster is more complicated, it can store a large amount of data and has less impact on Redis’s main thread blocking.

A sharding cluster, also known as a partitioned cluster, refers to the composition of multiple Redis instances into a cluster, and then dividing the received data into multiple portions according to certain rules, with each portion being stored by one instance. Returning to our previous scenario, if we evenly divide the 25GB data into 5 portions (of course, uneven division is also possible) and use 5 instances to store them, each instance only needs to store 5GB of data. The figure below illustrates this:

Architecture diagram of a sharding cluster

In a sharding cluster, when instances generate RDB files for 5GB of data, the data volume is much smaller, and the fork operation is generally unlikely to cause a long blocking time for the main thread. By using multiple instances to store data shards, we can both store 25GB of data and avoid the sudden slowdown of response caused by the fork operation blocking the main thread.

In practice, when using Redis, it is often inevitable to encounter situations where a large amount of data needs to be stored as the number of users or business scales expand. And a sharding cluster is a very good solution. In this lesson, let’s learn more about it.

How to store more data? #

In the previous case, we used two methods, namely, a high-memory cloud server and a sharded cluster, to store a large amount of data. In fact, these two methods correspond to two solutions for handling increased data volume in Redis: vertical scaling (scale up) and horizontal scaling (scale out).

• Vertical Scaling: Upgrading the resource configuration of a single Redis instance, including increasing memory capacity, increasing disk capacity, and using a higher CPU configuration. As shown in the figure below, the original instance had 8GB of memory and 50GB of disk space, and after vertical scaling, the memory increased to 24GB and the disk space increased to 150GB.
• Horizontal Scaling: Increasing the number of current Redis instances horizontally. As shown in the figure below, the original setup used 1 instance with 8GB of memory and 50GB of disk space, and now it uses three instances with the same configuration.

Comparison of vertical scaling and horizontal scaling

So, what are the advantages and disadvantages of these two approaches?

Firstly, the benefit of vertical scaling is that it is simple and straightforward to implement. However, this approach also faces two potential issues.

The first issue is that when using RDB for data persistence, if the data volume increases, the required memory will also increase, and the main thread may be blocked when forking child processes (as in the previous example). However, if you do not require persistent storage of Redis data, then vertical scaling can be a good choice.

However, in this case, you will also face the second issue: vertical scaling is limited by hardware and cost. This is easy to understand, as expanding memory from 32GB to 64GB is relatively easy, but expanding it to 1TB will face hardware capacity and cost limitations.

Compared to vertical scaling, horizontal scaling is a more scalable solution. This is because, to store more data with this approach, you only need to increase the number of Redis instances, without worrying about the hardware and cost limitations of individual instances. When dealing with millions or tens of millions of users, a Redis sharded cluster using horizontal scaling will be a great choice.

However, when using a single instance, it is very clear where the data exists and where the client accesses it. In contrast, a sharded cluster inevitably involves distributed management of multiple instances. To make use of a sharded cluster, we need to address two major issues:

• How to distribute data among multiple instances after sharding?
• How can the client determine which instance to access for the desired data?

Next, we will address these problems one by one.

Correspondence Between Data Slices and Instances #

In a sharded cluster, data needs to be distributed across different instances. So how do data and instances correspond to each other? This is related to the Redis Cluster solution that I’m going to talk about next. However, we first need to understand the relationship and differences between sharded clusters and Redis Cluster.

In fact, a sharded cluster is a general mechanism for storing a large amount of data, and this mechanism can have different implementation approaches. Before Redis 3.0, the official documentation did not provide a specific solution for sharded clusters. Starting from version 3.0, the official Redis cluster solution, called Redis Cluster, was introduced to implement sharded clusters. The Redis Cluster solution specifies the rules for data and instance correspondence.

Specifically, the Redis Cluster solution uses hash slots (referred to as “slots” from now on) to handle the mapping between data and instances. In the Redis Cluster solution, a sharded cluster has a total of 16384 hash slots, which are similar to data partitions. Each key-value pair will be mapped to a hash slot based on its key.

The mapping process consists of two major steps: Firstly, a 16-bit value is calculated based on the key of the key-value pair using the CRC16 algorithm; Secondly, this 16-bit value is then used to calculate the modulus of 16384, resulting in a modulus in the range of 0 to 16383, where each modulus represents a corresponding numbered hash slot. The CRC16 algorithm is not the focus of this lesson, so you can briefly review the information in the link.

So how are these hash slots mapped to specific Redis instances?

When deploying the Redis Cluster solution, we can use the cluster create command to create a cluster, and Redis will automatically distribute these slots evenly among the cluster instances. For example, if there are N instances in the cluster, then the number of slots per instance will be 16384/N.

Of course, we can also manually establish connections between instances using the cluster meet command to form a cluster, and then use the cluster addslots command to specify the number of hash slots on each instance.

Let’s take an example. Suppose the Redis instances in the cluster have different memory size configurations. If the hash slots are evenly distributed among all instances, when saving the same number of key-value pairs, the instances with smaller memory sizes will have greater capacity pressure compared to instances with larger memory sizes. In such cases, you can use the cluster addslots command to manually allocate hash slots based on the resource configuration of each instance.

To help you understand, I will provide a diagram to explain the mapping and distribution of data, hash slots, and instances.

In the diagram, the sharded cluster has a total of 3 instances, and let’s assume there are 5 hash slots. We can manually allocate hash slots using the following commands: instance 1 stores hash slots 0 and 1, instance 2 stores hash slots 2 and 3, and instance 3 stores hash slot 4.

redis-cli -h 172.16.19.3 –p 6379 cluster addslots 0,1
redis-cli -h 172.16.19.4 –p 6379 cluster addslots 2,3
redis-cli -h 172.16.19.5 –p 6379 cluster addslots 4

During the operation of the cluster, after the CRC16 value of key1 and key2 is calculated and modulo is performed with the total number of hash slots, the respective modulus results will determine which instance key1 and key2 are mapped to: instance 1 and instance 3.

Furthermore, I want to remind you that when manually allocating hash slots, it is necessary to allocate all 16384 slots, otherwise the Redis cluster will not work properly.

Alright, with the help of hash slots, the sharded cluster achieves the distribution of data to hash slots and then to instances. But how do client applications know which instance to access for specific data? Let’s talk about that next.

How does the client locate data? #

When locating key-value pair data, the hash slot it resides in can be calculated and executed on the client when sending a request. However, to further locate the instance, it is necessary to know on which instance the hash slots are distributed.

Generally, after the client establishes a connection with the cluster instances, the instances will send the hash slot allocation information to the client. However, when the cluster is just created, each instance only knows which hash slots it is allocated, and does not know the hash slot information owned by other instances.

So how can the client obtain all the hash slot information when accessing any instance? This is because Redis instances will send their hash slot information to other connected instances to complete the spreading of hash slot allocation information. After the instances are connected to each other, each instance will have the mapping relationship of all hash slots.

After receiving the hash slot information, the client caches the hash slot information locally. When the client requests a key-value pair, it first calculates the hash slot corresponding to the key, and then can send the request to the corresponding instance.

However, in a cluster, the relationship between instances and hash slots is not immutable, and the most common changes are:

• In the cluster, instances are added or deleted, and Redis needs to reassign hash slots.
• In order to achieve load balancing, Redis needs to redistribute hash slots on all instances.

At this time, instances can still obtain the latest hash slot allocation information by exchanging messages with each other. However, the client cannot actively perceive these changes. This will result in the inconsistency between the cached allocation information and the latest allocation information. So, what should be done in this case?

Redis Cluster provides a redirection mechanism, which means that when the client sends a data read or write operation to an instance that does not have the corresponding data, the client needs to send the operation command to a new instance.

So how does the client know the access address of the new instance during redirection? When the client sends a key-value pair operation request to an instance and there is no hash slot mapping for this key-value pair on this instance, the instance will return the following MOVED command response to the client, which includes the access address of the new instance.

GET hello:key
(error) MOVED 13320 172.16.19.5:6379

In this response, the MOVED command indicates that the hash slot 13320 of the requested key-value pair is actually located on the instance 172.16.19.5. By returning the MOVED command, the information of the new instance where the hash slot is located is provided to the client. In this way, the client can directly connect to 172.16.19.5 and send the operation request.

Let me illustrate the usage of the MOVED redirection command with a diagram. As can be seen, due to load balancing, the data in Slot 2 has been migrated from instance 2 to instance 3. However, the client cache still records the information “Slot 2 is on instance 2”, so it will send commands to instance 2. Instance 2 returns a MOVED command to the client, indicating the new location of Slot 2 (which is on instance 3), and the client will send the request to instance 3 again, while also updating the local cache to update the mapping relationship between Slot 2 and the instance.

It is worth noting that in the above diagram, when the client sends a command to instance 2, all the data in Slot 2 has been migrated to instance 3. In practical application, if there is a lot of data in Slot 2, there may be a situation where the data in Slot 2 is only partially migrated to instance 3, and some data is still in instance 2. In this case, when the migration of part of the data is completed, the client will receive an ASK error message, as shown below:

GET hello:key
(error) ASK 13320 172.16.19.5:6379

The ASK command in this result indicates that the hash slot 13320 of the requested key-value pair is on the instance 172.16.19.5, but this hash slot is currently being migrated. In this case, the client needs to first send an ASKING command to the instance 172.16.19.5, which means allowing this instance to execute the commands to be sent by the client next. Then, the client sends the GET command to this instance to read the data.

It may seem a bit complicated, let me explain with the help of an image.

In the following image, Slot 2 is being migrated from instance 2 to instance 3, where key1 and key2 have already been migrated, while key3 and key4 are still on instance 2. When the client requests key2 from instance 2, it will receive an ASK command from instance 2.

The ASK command has two meanings: first, it indicates that the Slot data is still being migrated; secondly, the ASK command returns the access address of the new instance where the requested data resides to the client. At this time, the client needs to send an ASKING command to instance 3 and then send the operation command.

Unlike the MOVED command, the ASK command does not update the hash slot allocation information cached by the client. Therefore, in the above diagram, if the client requests data in Slot 2 again, it will still send the request to instance 2. This means that the role of the ASK command is only to allow the client to send a request to the new instance once, and it does not change the local cache like the MOVED command, so that all subsequent commands will be sent to the new instance.

Summary #

In this lesson, we learned about the advantages of using a sharded cluster to store large amounts of data, as well as the data distribution mechanism based on hash slots and the method of locating key-value pairs for clients.

When dealing with data expansion, although increasing memory through vertical scaling is a simple and direct method, it can cause the database to have excessive memory, resulting in slower performance. Redis sharded cluster provides a horizontal scaling mode, which means using multiple instances and configuring a certain number of hash slots for each instance. Data can be mapped to hash slots through the hash value of the key and then scattered and stored on different instances. The benefit of doing so is that the sharded cluster can handle any amount of data.

In addition, instance scaling in a cluster or data redistribution for load balancing may change the mapping relationship between hash slots and instances, resulting in error messages when clients send requests. By understanding the MOVED and ASK commands, you won’t be troubled by these error messages.

I just mentioned that before Redis 3.0, the Redis official did not provide a sharded cluster solution. However, at that time, the industry already had some sharded cluster solutions, such as ShardedJedis based on client partitioning, and proxy-based solutions like Codis and Twemproxy. These solutions were implemented earlier than the Redis Cluster solution and have their own advantages in terms of supported cluster instance scale, cluster stability, and client-friendliness. I will discuss the implementation mechanisms and practical experiences of these solutions specifically in the upcoming lessons. This way, when you encounter the challenge of dealing with huge amounts of data brought by business development, you can choose the appropriate solution based on the characteristics of these solutions to implement a sharded cluster that meets your business needs.

One Question for Each Lesson #

As usual, I have a small question for you: The Redis Cluster solution distributes key-value pairs to different instances using hash slots. This process requires calculating the CRC of the key-value pair’s key and then mapping it to a hash slot. What are the benefits of doing this? If we directly record the correspondence between key-value pairs and instances in a table (for example, key-value pair 1 is on instance 2, key-value pair 2 is on instance 1), we wouldn’t need to calculate the correspondence between the key and hash slots. We would only need to look it up in the table. Why doesn’t Redis do it this way?

Feel free to express your thoughts in the comments section. If you find this valuable, I hope you can share today’s content with your friends and help more people solve sharding cluster issues.