Partitioning (sharding) in plain words: slicing data across nodes without a hot spot

← Back to the section

When there's more data or load than a single server can handle, you slice it into pieces — partitions (aka shards). The rule is simple: each record lives in exactly one partition, and the partitions are spread across different nodes. The practice of sharding we cover on MongoDB — replica set, shard key, chunks, balancer. This article is one level up and not tied to a database: the same questions come up the same way in MongoDB, Cassandra, Elasticsearch, and HBase — only the names change.

The goal is evenness; the enemy is the hot spot

The point of partitioning is one thing: spread the data and the load across nodes evenly, so that ten nodes handle ten times as much as one. If the split comes out uneven — some partitions got a disproportionate amount of data or requests — it's called skewed, and an overloaded partition is a hot spot. In the worst case all the load lands on one node while the other nine sit idle — and the whole point of sharding is lost.

There are two basic ways to decide which partition a record goes to — by key range and by key hash.

Range partitioning versus hash partitioning

By key range. Each partition gets a continuous stretch of keys — like the volumes of a paper encyclopedia: A–C, D–F, and so on. The boundaries are chosen to fit the data, because the distribution is uneven (there are different numbers of words starting with "A" and with "X"). The upside of this split is range queries: within a partition the keys are in order, so "all events for March" reads as one fast scan. The downside is that the same order easily breeds a hot spot. If a key starts with a timestamp, then all of today's writes land in one partition (today's range) while yesterday's nodes stand idle. Cured by a composite key where time isn't first: say, the sensor name first and only then the time — then writes for one moment spread across different sensors.

By key hash. A good hash function turns even similar keys into evenly scattered numbers, and a partition gets a range not of the keys themselves but of their hashes. Evenness comes almost for free — that's what Cassandra and MongoDB do. The price is losing the sorting: keys that are neighbors in meaning fly off to all partitions, and a range query now has to poll every node. Cassandra takes a compromise — a composite key: only the first part is hashed (it determines the partition), while the rest work as a sorted index within the partition. Then "all of one user's messages over a period" reads efficiently, while the users themselves are spread evenly across the cluster.

There's a separate trouble that neither hash nor range fixes — the hot key. For example, a celebrity with millions of followers: all the traffic hits one of their records, and the hash of the same key is always the same, so you can't "spread it out" automatically. Only the application helps here: glue a random suffix of a hundred variants onto the key — then the write spreads across a hundred partitions. But reads pay for it: now, to gather everything under that key, you have to poll all hundred partitions and stitch the results. So the trick is applied to a handful of known hot keys, not to everything.

Secondary indexes: local versus global

As long as we access data by primary key, everything is simple: it determines the partition. But the application also needs secondary indexes: "find all red cars," "all articles with the word hogwash." A secondary index doesn't line up with the partitioning, and there are two ways to slice it — with diametrically opposite costs.

Local index (document-partitioned). Each partition keeps a secondary index only over its own documents. You put a red car in partition 3 — the entry "color: red" appears in partition 3's index and nowhere else. The write is cheap: everything is local, one partition is touched. But the read is scatter/gather: red cars are scattered across all partitions, so the query "find red ones" has to be sent to every node and the answers stitched together. That's how MongoDB, Cassandra, Elasticsearch work. Scattered reads lengthen the latency "tail" (you wait for the slowest node of all), but the write stays local — which is why this is the default.

Global index (term-partitioned). Here there's one index for the whole cluster, but it too is sliced into partitions — by the value being searched. The whole entry "color: red" sits in one partition of the index. The read is fast — you go to one partition instead of all. But the price moves to the write: adding one document with several fields touches several index partitions at once (the values "color" and "make" live in different places), and keeping that consistent on the fly is a distributed transaction. So global indexes are updated asynchronously: after a write, the change appears in the index with a delay (that's how DynamoDB works — "usually a fraction of a second, longer during faults").

Rebalancing: how to move partitions between nodes

Over time nodes are added (load grew) and removed (a fault or decommission). Moving partitions from one node to another is called rebalancing, and you expect three things from it: afterward the load is even again; during it the database keeps reading and writing; no more data is moved than actually necessary.

• How not to do it — hash mod N. It's tempting to assign a partition by the formula hash(key) % N, where N is the number of nodes. But change N, and almost all keys change partition: going from 10 to 11 nodes reshuffles practically all the data. Rebalancing becomes unbearable.
• A fixed number of partitions. You create far more partitions than nodes (say, 1000 partitions across 10 nodes) and hand out many per node. A new node simply "takes over" a few partitions from the existing ones. Whole partitions move; their count and the key assignment don't change — only which node holds which partition changes. That's how Elasticsearch, Riak, Couchbase do it. The pitfall: the number of partitions has to be guessed in advance — it sets the ceiling for growth, and if chosen too large it brings extra overhead.
• A dynamic number of partitions. A partition outgrew a threshold — it splits in two; it shrank — it merges with a neighbor (just like a node in a B-tree). The number of partitions adjusts itself to the volume of data. That's how HBase, MongoDB (chunks), RethinkDB work. The empty-database pitfall: at the start there's one partition, and all writes fly to one node until the first split happens (cured by pre-splitting, if the distribution is known ahead of time).

"Almost all" is arithmetic, not a figure of speech. Count how many of a hundred thousand keys change nodes when going from ten nodes to eleven — first by hash mod N, then with a thousand fixed partitions handed out whole.

live example

public class Rebalance {
    public static void main(String[] args) {
        int keys = 100000;
        int partitions = 1000;
        int target = partitions / 11;

        int movedByMod = 0;
        for (int i = 0; i < keys; i++) {
            if (hash(i) % 10 != hash(i) % 11) movedByMod++;
        }

        int[] load = new int[10];
        for (int p = 0; p < partitions; p++) load[p % 10]++;
        boolean[] moved = new boolean[partitions];
        for (int p = 0, taken = 0; p < partitions && taken < target; p++) {
            if (load[p % 10] > target) { load[p % 10]--; moved[p] = true; taken++; }
        }

        int movedByPartition = 0;
        for (int i = 0; i < keys; i++) {
            if (moved[hash(i) % partitions]) movedByPartition++;
        }

        System.out.println("hash mod N:            " + Math.round(100f * movedByMod / keys) + "% of keys move");
        System.out.println("1000 fixed partitions: " + Math.round(100f * movedByPartition / keys) + "% of keys move");
    }

    static int hash(int key) {
        return ("order-" + key).hashCode() & 0x7fffffff;
    }
}

Ninety-one percent against nine: fixed partitions move exactly the share the new node needs — about one eleventh.

And a separate question — automatic versus manual control. Fully automatic rebalancing is convenient but dangerous paired with automatic fault detection: an overloaded node responds slowly → the system decides it "failed" → starts a rebalance → adds load to the already-overloaded node → a cascading failure. A human in the loop ("the system proposes, an admin confirms," as in Couchbase and Riak) is slower but saves you from such surprises.

Request routing: where does "foo" live?

Partitions are spread across nodes and move during rebalancing. How does a client know which node to go to for the key foo? This is a special case of the general "find who's responsible for what" problem (service discovery), and there are three solutions:

1. The client knocks on any node. If the key is there — the node answers itself; if not — it forwards the request to the right node itself and returns the answer. That's how Cassandra and Riak do it, via a gossip protocol: the nodes exchange the cluster map among themselves, and no external coordinator is needed.
2. A separate routing tier. All requests go through it, and it knows which node is responsible for what (in MongoDB that's mongos). It doesn't process the requests itself — it's a load balancer that simply knows where each partition is.
3. The client itself knows the layout and connects to the right node directly, without intermediaries.

In all three cases someone has to know the current "partition → node" map and learn about changes in time. The classic solution is a separate coordinator service, ZooKeeper: nodes register in it, routers subscribe to changes, and when a partition moves, ZooKeeper notifies everyone. That's how HBase and SolrCloud do it. Kafka used to as well, but since 3.3 it can keep the map itself, and 4.0 dropped the ZooKeeper dependency entirely — that's the general trend: an external coordinator is one more system to run. The alternative is that same gossip (Cassandra, Riak): the nodes are more complex, but there's no dependency on an external coordinator.

Where this applies

The "how to shard" fork comes up exactly when one node stops coping. And nearly every decision here is irreversible: the shard key is very hard to change later, and you can't re-do the index scheme after launch.

Where beginners stumble:

• The partition key starts with a time or an auto-increment — and all fresh writes land in one partition. The classic hot spot. The first part of the key needs something varied.
• Expecting a secondary index to be as fast as the primary — but with local indexes any query not by the partition key turns into a scatter/gather across all nodes.
• Forgetting about the hot key — one celebrity or one popular product takes down a partition, even though the distribution formally looks even.

In short

• The point of partitioning is to spread both the volume and the load across nodes. A skew gives a hot spot, and the whole point of sharding is lost.
• Range partitioning gives fast scans in key order, but a time prefix drives the whole fresh stream into one partition. A composite key cures it.
• Hash partitioning gives evenness almost for free, and the sorting is what you pay with. A hash does not cure a hot key: the application spreads it with a suffix, and reads pay for that.
• A local secondary index means cheap writes and scatter/gather on reads — the default. A global one is the reverse and is updated asynchronously.
• hash mod N is unusable: going from 10 nodes to 11 moves 91% of the keys, a thousand fixed partitions move 9%.
• The "partition → node" map is kept either by an external coordinator (ZooKeeper) or by the nodes themselves via gossip. Fully automatic rebalancing is not paired with automatic fault detection.

What to read next

• MongoDB replication and sharding — shard key, chunks, balancer, and mongos in practice.
• Replication models — partitioning almost always goes together with replication.
• Partitioning and sharding in PostgreSQL — RANGE, LIST, and HASH inside one database.
• The building blocks — where sharding lives in the bigger picture.