Sharding: splitting data across servers

Sometimes the biggest database server is not enough. The writes are too many for it, or the data does not fit on its disk. Copies of the data do not fix this. So interviewers ask: "The data and the writes are too big for one server. What do you do?"

A weak answer says "use sharding" or "use NoSQL" and stops. A strong answer says how to split the data over several servers, and which column decides where each row goes. It also says which questions become slower, and how to add a server later.

At SDE-1 level you should explain why one server is not enough and what splitting means. At SDE-2 level you should choose the column to split on and give a reason. At SDE-3 level you should also say what happens to queries, transactions and the data when you change the number of servers.

A school has 4,000 students, and one register is too big. One clerk cannot write all the changes. Copies of the register in each classroom (lesson 39) do not help. Every change still goes to one master register, and every copy is as big as the master. So the school splits the students into four registers, each with its own clerk. Now a clerk handles a quarter of the changes and keeps a quarter of the students.

But now each student must be in exactly one register, and you must know which. The school needs a rule. It could say "classes 1 to 3 are in register A, classes 4 to 6 in register B" and so on. It could say "divide the roll number by 4 and use the remainder". Or it could keep a notice board that lists the register of each student. These are the three ways you will meet below.

Sharding means splitting the data of one database over several database servers, so that each server holds only a part. Each part is a shard. A shard is a small, complete database. It has its own leader, and it can have its own followers (lesson 39). The shard key is the column that decides which shard owns a row. In lesson 17 a wide-column store split rows by their partition key. The shard key does the same job.

Sharding solves the two limits that followers cannot: the number of writes, and the size of the data. Each shard takes only its part of both.

To find a row, the app uses the key. This is called routing. A query that contains the shard key goes to one shard. A query that does not contain the key must ask every shard and join the answers. This is a scatter-gather query. It does the work on every shard, and it waits for the slowest one.

Three ways to choose the shard

Range. Each shard owns a range of key values, such as orders from January, February, March and April. It is good for queries over a range ("all orders from 1 to 7 October"), because they touch few shards. It is bad when the key only grows, such as the date. All new rows go to the newest shard, and the others do almost nothing.

Hash. A hash function turns any value into a number. The same value always gives the same number, and different values give numbers that look random. You write "x mod 4" for the remainder when you divide x by 4: it is 0, 1, 2 or 3. The shard is the hash of the key mod the number of shards. Keys spread evenly. The cost is that keys next to each other are not together, so a range query must ask every shard. A second cost is that changing the number of shards moves most rows.

Directory. A table lists the shard of each key, such as "user 101 is on shard 2". The app reads the table to route. You can place any key anywhere and move one key at a time. The costs are an extra lookup, and one more part that must be correct, fast and always available.

How to choose the key

1. It is in the most common queries, so most queries ask one shard.
2. It has many different values that spread evenly. A key with few values, such as city, gives uneven shards. One big city can fill a shard while a small town's shard stays almost empty.
3. It does not only grow, like a date or a counter.
4. Rows that change together share it, so one transaction stays on one shard.

What sharding costs

• Queries without the key ask every shard.
• Transactions and joins across shards are hard and slow. An all-or-nothing change over two servers is hard (lesson 21). A join combines rows of two tables (lesson 17). Keep rows that change together, or are joined, on one shard.
• Unique IDs. A counter on each shard gives the same number twice. Use an ID that is unique across shards (lesson 23).
• Servers. Each shard has its own leader and followers, so the server count multiplies.

Adding a shard

Suppose you go from 4 to 5 shards with a hash mod. A row stays only if its hash mod 4 and its hash mod 5 are the same number. That is true for only 1 row in 5. So 4 of every 5 rows move to a different shard. For N shards going to N + 1, a share N / (N + 1) moves. Going from 4 to 8 moves half. With a directory, you move only the keys you choose. With ranges, you split one range in two. To move rows while the app runs:

1. Copy the rows to the new shard while the old shard still serves them.
2. Keep sending new changes to the new shard, as a follower does (lesson 39).
3. Switch the routing for those keys.
4. Delete the old copy.

The copying uses network and CPU on both servers.

SwadKart's orders database has grown. We decide how many shards it needs, and which key to use.

Assumptions (made up to teach with, not real traffic):

• At peak there are 5,000 writes per second. A write costs 8 ms of CPU.
• A server has 16 cores, which is 16,000 ms per second (lesson 15). We use at most 70% of it: 11,200 ms.
• The data is 9 TB. A server disk holds 4 TB, and we fill it to 70% at most: 2.8 TB.
• 80% of reads are "my orders" (by user id), 15% are "one order" (by order number), 5% are the restaurant's page (by restaurant id).
• A query that asks a shard does the same work on that shard whether or not the shard has the row.
• An order number does not contain the user or the date.

Step 1: one server is not enough. Writes = 5,000 x 8 = 40,000 ms per second. That is 40,000 / 16,000 = 250% of the server. The data is 9 TB on a 4 TB disk. Followers do not help: they take no writes, and each holds all 9 TB.

Step 2: how many shards.

• By writes: 40,000 / 11,200 = 3.6, so 4 shards.
• By data: 9 / 2.8 = 3.2, so 4 shards.
• With 4 shards each takes 40,000 / 4 = 10,000 ms (62.5% of a server) and 9 / 4 = 2.25 TB of data.

Step 3: the key. We count how many shards the average read asks. The fractions are the shares of reads from the assumptions.

User id wins. Reads cost 1.6 times the work of reads that ask one shard, and a user's orders sit together. The date range also has a write problem: all 5,000 writes per second go to the newest shard, which is 250% busy. Hashing by user id gives each shard 62.5%.

This answers the predict box. With the order number only, you must ask all 4 shards, because the key is the user id.

Step 4: routing with a hash. Assumption: we use a hash function called CRC32 on the text "user:101". The shard is the hash mod 4.

Users 101 and 103 are on shard 2. Users 102 and 104 are on shards 0 and 1.

Step 5: a fifth shard. The "mod 5" column shows where users go with 5 shards. All four users change shard. On average 4 / 5 = 80% of rows move.

• Hash mod: 0.8 x 9 TB = 7.2 TB moves. Assume one network link of 1 Gbps does all the copying. It carries 125 MB per second (lesson 15). 7,200,000 MB / 125 = 57,600 seconds = 16 hours.
• A directory moves only what the new shard needs: 9 / 5 = 1.8 TB, which is 20%. That takes 4 hours.

Step 6: servers. Before: 1 leader and 2 followers = 3 servers. Now: 4 shards x 3 = 12 servers.

• It solves writes and data that one server cannot hold. In the example, 4 shards took 250% of a server to 62.5% each.
• It costs slower queries without the key (1.6 times the work, instead of 1.0). It makes transactions and joins harder. It needs 12 servers instead of 3. Changing the number of shards moves a lot of data (7.2 TB with hash mod).
• I would change it when the cheaper steps are used up. Those are indexes (sorted lists that find rows fast, lesson 20), a cache (a fast store of popular data, lesson 31) and followers (lesson 39). Also try a bigger server and removing old data. I would also change it when the key stops fitting the queries.

"Sharding is the first fix when the database is slow." It comes after cheaper fixes. In the example it brought 1.6 times the read work and 4 times the servers. Check indexes, a cache, followers and a bigger server first.

"Sharding by date is good, because queries use dates." Only for reading. The key only grows, so all new writes land on one shard. In the example that shard was 250% busy while three shards did almost nothing.

"With hash sharding I can add one shard when I need it." The move is very large. Going from 4 to 5 shards moved 80% of 9 TB, which is 16 hours of copying.

1. What is a shard, and what is a shard key? Name three ways to decide which shard owns a row.
2. A table has 6 TB of data and 3,000 writes per second at 8 ms each. A server has 16 cores (use 70%) and a 4 TB disk (fill to 70%). How many shards do you need for writes, and how many for data?
3. A team shards a chat database by message date, one shard per month. New messages are the most read and written. What goes wrong, and what key would you suggest?