The labs on this site need JavaScript to run. The same content is written out below.
Learn › Database › Scaling and distribution
Sharding
What happens when you split data across several databases
When one table grows too big for one database server, the rows are divided among several databases (shards). The shard key and the rule for dividing decide which row goes to which shard.
Checked against real execution · The same data was calculated separately in Java and checked against the results
Concepts in this lab
- Sharding and partitioning
- Partitioning divides a large table into several pieces by a criterion such as a date range and manages it that way. Sharding stores data across several database servers. It spreads out a write load or a data volume that one server cannot handle. With sharding, joins and transactions that span servers become difficult.
- Scale up and scale out
- Scaling up means raising the specifications (CPU, memory) of a single server to increase its capacity, and scaling out means adding more servers to share the load. Scaling up is simple but has limits and cost problems. Scaling out can grow flexibly, but the servers must be designed to hold no state, and you need something that distributes the requests.
- Replication
- Continuously copying the changes of an original database (the source, also called the primary) to another database (a replica) so that both keep the same data. It reduces load by spreading read requests across replicas, and if the source fails, a replica can be promoted to take its place and keep the service running. There is a small delay in the copying, so data that was just written may not be visible on a replica right away.
Try these
- Default settings — Orders are distributed by the remainder of the member number divided by 4. They spread evenly.
- Dividing by order number range — All new orders go to the last shard.
- When a particular member has many orders — Only the shard that member belongs to grows.
- Adding a shard with hashing — From 4 shards to 5. Most of the data has to move.
- Adding a shard with ranges — The existing data stays where it is, and the new shard receives only the new range.
Running with the default settings
- Divide 120 orders among 4 shards. The remainder of the member number of the order divided by the number of shards (4) is the shard number. The orders of the same member always end up on the same shard.
- Distribution result. Largest shard: 30 orders. Smallest shard: 30 orders. The member numbers are spread evenly, so the number of orders per shard is similar.
- New orders are scattered over several shards depending on the member number of the member who ordered. The write load is spread out.
- "Orders of member 7" query: 7 % 4 = 3. Only that shard needs to be looked at. Conversely, a query that finds orders without a member number, such as "all orders from yesterday", has to query every shard.
Result of this run
- Once chosen, a shard key is very hard to change. The most frequently used query condition should be the shard key, so that a query needs to look at only one shard. Queries without the shard key, and joins and transactions that span shards, become difficult.
- Sharding is the last resort for dividing writes and data volume. Before it, first consider improving indexes and queries, caching, read-only replicas, and separating old data.
Code
// The shard is chosen by the remainder of the member number divided by the number of shards
int shardOf(long memberId) {
return (int) (memberId % 4);
}
// The same calculation finds the shard when querying
DataSource ds = shards.get(shardOf(memberId));
Assumptions of this model
This is a small example with 120 orders and 40 members. Member numbers were assigned evenly to the orders, and with the "many orders" setting, one in every three orders was given to member 7. The ratio of data that moves when the number of shards changes is a value actually calculated from this data. A setup with replicas for each shard and the router that manages shard locations are left out. For 2 to 6 shards, the number of orders that must move when one shard is added was calculated separately in Java, and it matched the values on the screen.
Previous lab: Replication lag and CAP · Next lab: Partitioning