Module 5: Scaling the Database

5. Sharding and the shard key

Next →

Description

By the end of this lesson you'll understand the second horizontal-scaling tool and the one that attacks the axis replicas can't: sharding, splitting the data into pieces (shards) and storing each piece on a different machine. You'll see why you shard —when neither Enlace's ~6 TB nor its writes fit on a single machine—, what the shard key is (the field that decides which shard each record lives in), why in Enlace that key is the short_code, and the central distinction between partitioning by range and by hash. And you'll see, by running code, the phenomenon that makes or breaks a sharding plan: the hotspots, when a bad key concentrates almost all the load on a single shard while the others sit idle.

This matters because sharding is the hardest decision to reverse in all of scaling. Adding a replica is reversible and cheap; choosing the shard key badly condemns you to a system where one node burns while the others sleep, and rebalancing it later —with the data already spread and the traffic live— is one of the most painful operations that exist. The shard key is one of those decisions that, made well at the start, you don't think about again for years, and made badly, chases you all that time. This lesson teaches you to make it with judgment, and it prepares the ground for the two that finish it off: how to spread by hash without the mod-N trap (lesson 6) and how consistent hashing disarms it (lesson 7).

Connection to the module: this lesson crosses from the reads axis (replicas, lessons 3-4) to the data and writes axis (sharding). It's the one that resolves limits 1 and 3 from lesson 2 —storage and write throughput—, the ones replicas left intact. And it's the first half of a three-lesson arc: here you establish what sharding is and why the key matters; lesson 6 shows why the obvious way to spread by hash (hash(key) % N) is a trap when growing; and lesson 7 resolves it with consistent hashing. Think of these three lessons as a single story told in three acts, and this is the setup.

The library that no longer fits in one building

Think of it this way. A library grows until its books no longer fit in a single building. The solution isn't an endlessly bigger building —that's vertical scaling, and it has a ceiling—: it's spreading the collection over several branches. Each branch stores a part of the books; among them all, the complete collection. Now the library can grow by adding branches, and each branch serves its own visitors in parallel.

But immediately the question that decides everything arises: by what criterion do you spread the books among branches? It's the shard key, and it's not a detail. Look at two criteria:

Bad criterion — by acquisition date. Branch 1 stores the oldest books, branch 4 the newest. It sounds orderly, but it has a fatal defect: all the new books that arrive each day go to the same branch —the "most recent" one—. That branch lives overwhelmed with work while the other three, with the old books almost no one touches, are idle. You concentrated the work in one point. That's a hotspot.

Good criterion — by a code that spreads evenly. You assign each book to a branch with a rule that doesn't correlate with when it arrived or how popular it is —for example, a hash of its catalog code—. Today's new books are spread among the four branches equally, and so are the old and the popular ones. No branch is overwhelmed; the work is distributed. There's no hotspot.

The library's lesson is sharding's lesson: spreading is easy; spreading evenly is the art. The spreading criterion —the shard key— decides whether your machines share the load or whether one burns while the others sleep. In Enlace, the question is identical: by which field of the Link record do you decide which shard each link lives in?

What the shard key is, and why in Enlace it's the short_code

The shard key (or partition key) is the field of the record whose value decides which shard that record lives in. It's a function: shard = f(shard_key). Every operation that knows the key can go straight to the correct shard without asking the others; every operation that does not know it has to ask them all —a query that "crosses shards", slow and to be avoided—.

In Enlace, the dominant operation is resolve(short_code) -> long_url: a short_code arrives and its long_url has to be found. If you choose the short_code as the shard key, then resolve knows exactly which shard to go to —it computes f(short_code) and queries only that shard—. It's an almost perfect choice for Enlace for three reasons:

  1. The dominant read knows it. resolve always brings the short_code in hand; it never has to query all the shards. Each resolution touches exactly one shard.
  2. It's high-cardinality and unique. There are 62⁷ ≈ 3.5 trillion possible short_codes, all different; a key with a great many distinct values spreads finely, without clumps.
  3. It doesn't correlate with load. A short_code doesn't say whether the link will be popular or when it was created (if you hash it), so spreading by it doesn't concentrate the load —unlike spreading by date—.

And the write? shorten creates a Link with a new short_code; it also knows the key, so it knows which shard to write to. Both paths —read and write— know the short_code, and that's why it's Enlace's natural key.

By range vs. by hash

With the key chosen, it remains to decide how the key translates to a shard. There are two families, with opposite tradeoffs.

Partitioning by range. You assign contiguous ranges of the key to each shard: shard 0 stores the short_codes from 0000000 to Fffffff, shard 1 from G to V, and so on. Advantage: range queries are efficient ("give me all the codes between X and Y" touch few contiguous shards). Fatal disadvantage for sequential keys: if the short_codes are generated with an increasing counter (as we saw in module 3), all the new codes fall in the highest range, that is on the same shard —a write hotspot—. The fresh codes are the ones that receive the writes and the first reads; concentrating them on one shard kills the purpose of sharding.

Partitioning by hash. You apply a hash function to the key and the result decides the shard: shard = hash(short_code) % N (or consistent hashing, which is where we're headed). Advantage: the hash destroys the order and the correlation; consecutive codes, or ones created the same day, are dispersed equally among all the shards. There's no hotspot by sequence. Disadvantage: you lose efficient range queries (a range of codes ends up scattered across all the shards). For Enlace it's not a loss —no one asks for "all the codes between X and Y"; the operation is resolve of a single code—, so by hash is Enlace's choice.

Worked example: the hotspot, measured

Let's see the hotspot with numbers, not words. We take a batch of 100 new short_codes, generated by a consecutive counter (as in module 3), and we spread them among 4 shards the two ways: by range of the code and by hash of the code.

import hashlib
from collections import Counter

N = 4  # 4 shards

def make_sequential_codes(k, start):
    """short_codes from a counter -> base62. Consecutive, as in module 3."""
    alphabet = "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"
    out = []
    for i in range(k):
        num, s = start + i, ""
        while num:
            s, num = alphabet[num % 62] + s, num // 62
        out.append(s.rjust(7, "0"))
    return out

new_batch = make_sequential_codes(100, start=500_000_000)  # 100 just-created codes

def shard_by_range(code):        # range by the code's first character
    first = code[0]
    if first <= 'F': return 0
    if first <= 'V': return 1
    if first <= 'l': return 2
    return 3

def shard_by_hash(code):         # hash of the full code
    return int(hashlib.md5(code.encode()).hexdigest(), 16) % N

range_dist = Counter(shard_by_range(c) for c in new_batch)
hash_dist = Counter(shard_by_hash(c) for c in new_batch)

print("A) by RANGE of the code:", [range_dist.get(s, 0) for s in range(N)])
print("B) by HASH of the code: ", [hash_dist.get(s, 0) for s in range(N)])

What to expect. When you run it:

A) by RANGE of the code: [100, 0, 0, 0]
B) by HASH of the code:  [27, 27, 28, 18]

There's the hotspot, measured and unambiguous. By range, the 100 new writes fall on shard 0 and the other three receive zero —because the consecutive codes share the same prefix and fall in the same range—. That shard 0 would receive all the creation traffic while the others sleep: you scaled to 4 machines and one does 100% of the work. By hash, the same 100 writes are spread 27/27/28/18 —almost even—, because the hash breaks the correlation between "new code" and "same shard". Four machines sharing the load, which is the whole point of sharding.

This is why Enlace shards the short_code by hash, not by range. And that decision opens exactly the question of the next two lessons: hash(short_code) % N spreads beautifully... until you change N.

The problem that peeks out: what about when you add a shard?

Hash spreading has an Achilles' heel we haven't seen yet and that dominates the rest of the module. The formula shard = hash(short_code) % N depends on N, the number of shards. It spreads perfectly as long as N doesn't change. But sharding is, by definition, something you do to be able to grow: sooner or later you're going to add a shard (from 4 to 5, from 8 to 9) because the data keeps growing. And at that instant, N changes, and with it hash(short_code) % N changes for almost every key —that is, almost all the links would have to move shards at once—.

That massive move —moving terabytes between machines while the system is live, with the caches going cold at once— is a storm that can bring down the system. Lesson 6 quantifies it with the experiment that measures exactly how many keys are remapped, and lesson 7 shows the technique —consistent hashing— that reduces it from "almost all" to "only the necessary ones". For now, keep the setup: you chose short_code by hash as the shard key, it's the correct choice, and you just uncovered the problem that makes real sharding hard.

Common mistakes

Choosing a shard key the dominant read doesn't know (design mistake). What happens: you shard by a field —say the user_id that created the link— but the dominant operation, resolve(short_code), doesn't bring that field, so each resolution has to ask all the shards which one has the code. Why it happens: the key is chosen thinking about how the data groups, not how it's queried. How to detect it: ask yourself "does the most frequent operation know the shard key?". If not, each of those operations is a query that crosses all the shards. How to fix it: choose the key the dominant read always brings in hand; for Enlace, it's the short_code.

Sharding a sequential key by range (hotspot mistake). What happens: you partition by range a field that grows over time —an autoincrement ID, a date, a base62 counter—, and all the new writes pile up in the last shard. Why it happens: the range seems natural and orderly, and the hotspot isn't seen until the traffic arrives. How to detect it: if one shard has 90% of the writes and the others almost none, and that shard is the "most recent" range one, it's the classic hotspot. You measured it: [100, 0, 0, 0]. How to fix it: for sequential keys, spread by hash, not by range; the hash breaks the correlation between "new" and "same shard".

Sharding when replicating was enough (sequence mistake, again). What happens: you shard to resolve a reads problem, carrying all the complexity of sharding when a replica would have resolved it more simply. Why it happens: "more machines" feels like the answer to any saturation. How to detect it: if your limit is read throughput and the writes and the data fit on one machine, you don't need sharding. How to fix it: remember the two axes —sharding is for data and writes (limits 1 and 3); reads are replicas (limit 2)—. Enlace shards for the 6 TB, not for the reads.

Exercises

Exercise 1 — Choose the key. For each candidate field of the Link record, say whether it would be a good or bad shard key for Enlace and why. (a) short_code. (b) created_at (the creation date). (c) long_url. (d) The user_id of who created the link.

See solution
  • (a) short_code — good, the best. The dominant read (resolve) always knows it, it's very high cardinality (62⁷ values) and, hashed, doesn't correlate with load. Each resolution touches a single shard. It's Enlace's choice.
  • (b) created_at — bad. It's sequential: all the new writes share a recent date and, by range, fall in the same shard (hotspot [100,0,0,0]). Furthermore, resolve doesn't bring the date, so each resolution would cross all the shards. Double failure.
  • (c) long_url — bad. resolve doesn't know it (it precisely seeks to find the long_url from the short_code), so it's useless for routing the dominant read. Each resolution would cross shards.
  • (d) user_id — bad for Enlace. resolve doesn't bring the user_id; finding a code would require asking all the shards. It could make sense in a system where the dominant operation is "give me all this user's links", but that's not Enlace's operation.

Exercise 2 — Predict the spread. You have 8 shards and a batch of 1,000 short_codes just created by a consecutive counter. (a) If you shard by range of the code, how do the 1,000 writes approximately spread? (b) If you shard by hash, how do they spread? (c) Which shard "burns" in each case?

See solution
  • (a) By range: the ~1,000 writes fall almost all on a single shard, the highest range one (that of the most recent codes), because the consecutive codes share a prefix. Approximately [1000, 0, 0, 0, 0, 0, 0, 0].
  • (b) By hash: the ~1,000 spread evenly among the 8 shards, ~125 each (with small variations), because the hash breaks the correlation with the sequence.
  • (c) Which burns: by range, that single most-recent-range shard burns while the other seven sleep —a hotspot that nullifies the sharding—. By hash, none burns: the load is distributed, which is the goal. It's the same story as the experiment [100,0,0,0] against [27,27,28,18], scaled to 8 shards and 1,000 writes.

Exercise 3 — The tradeoff of losing the ranges. Partitioning by hash spreads evenly but loses efficient range queries. (a) Give an example of a range query that would be slow with hash. (b) Explain why Enlace doesn't care about that loss. (c) Describe a hypothetical system where it would matter and by range would be the correct choice despite the hotspot risk.

See solution
  • (a) A slow range query with hash: "give me all the short_codes that start with aX" or "all created in March". With hash, those records are scattered across all the shards, so the query has to ask them all and gather the results —slow—.
  • (b) Enlace doesn't care because its dominant operation isn't a range one: it's resolve(short_code), a lookup of a single, exact code. Enlace never asks for "all the codes between X and Y"; it asks for "the long_url of this code". A point lookup with hash is optimal (one shard), and the range queries we lose don't exist in the real load.
  • (c) A system where the range would matter: a time-series or logs database where the dominant query is "give me all the events between 10:00 and 11:00 today". There, partitioning by time range makes that query efficient (it touches few contiguous shards), and the write hotspot on the "current" shard is accepted or mitigated in other ways (for example, adding a prefix that spreads within the range). The general criterion: choose range when the load is range queries; choose hash when it's point lookups, like Enlace.

Summary and next step

In this lesson you crossed to the axis replicas don't scale: sharding, splitting the data among several machines so the 6 TB fit and the writes are spread. With the multi-branch library you saw that spreading is easy but spreading evenly is the art, and that the spreading criterion —the shard key— decides everything. You established why Enlace's key is the short_code: the dominant read always knows it, it's very high cardinality, and hashed it doesn't correlate with load. You contrasted partitioning by range (good for range queries, fatal for sequential keys) against by hash (spreads evenly, loses the ranges), and you measured the hotspot: by range, [100,0,0,0]; by hash, [27,27,28,18]. And you uncovered the problem that dominates the rest of the module: hash(short_code) % N spreads perfectly until you change N.

Before moving on you should be able to: define the shard key and justify Enlace's; explain why a sequential key by range creates a hotspot; decide between range and hash according to the load; and anticipate why adding a shard with mod-N is a problem.

What comes next is quantifying that problem. In lesson 6 you'll run the experiment that measures, with a million keys, how many are remapped when you go from N to N+1 shards using hash(key) % N. The number —I'll preview it— is devastating: almost all. You'll understand why changing the modulo remaps the whole world, and why that turns "adding a shard" into a migration storm. It's act two of the story; act three, consistent hashing, resolves it.

Resources

← Previous4. Replication lag and reading from replicasNext →6. The `hash(key) % N` problem