learning.lab
Scale & Ops· 28/33

Distributed Tables

Sharding across nodes

Everything so far assumed one server. Sharding is how ClickHouse scales past what one machine's disk and CPU can hold: the table is split by row across multiple servers, and a special engine (confusingly also just called a table) knows how to talk to all of them at once.

Two tables, two jobs

On each shard, you create an ordinary MergeTree table — the "local" table, which actually stores rows. Separately, you create a Distributed table, typically on every node, which stores no data at all — it's a router that knows the cluster topology and a sharding expression.

-- on every shard: the real table
CREATE TABLE events_local
(
    user_id UInt64,
    event   String
)
ENGINE = MergeTree
ORDER BY user_id;

-- on every node: the router
CREATE TABLE events AS events_local
ENGINE = Distributed(my_cluster, default, events_local, cityHash64(user_id));
SELECT count() FROM events
Distributed table
query entry point (no data stored)
Shard 1local MergeTree
Shard 2local MergeTree
Shard 3local MergeTree
Merged result
partial results combined, sent to client

Query the Distributed table and it forwards the query to events_local on every shard in parallel, then merges the partial results — sums get summed, counts get summed, top-N results get re-sorted and truncated. From the client's perspective it's one table; underneath, it's N independent MergeTree tables each holding a slice of the rows.

The sharding key decides the slice

cityHash64(user_id) above means every row for a given user_id always lands on the same shard — useful when queries frequently filter or aggregate per user, since that work never needs to cross shards. A poor sharding key (or none — random distribution) is fine for full-table scans but forces more cross-shard coordination for anything that needs to group by the key you didn't shard on.

Common mistake
Inserting into a Distributed table by default routes each row to its shard synchronously as part of the insert, which is slow and fragile over the network. Production setups almost always insert directly into the local table on each shard from the ingestion pipeline, or enable distributed_foreground_insert /async settings deliberately, rather than relying on default distributed inserts at scale.
Why it exists
Sharding solves a different problem than replication. Sharding is about capacity — spreading data too big for one machine across many. Replication is about availability — keeping copies so no single machine failing loses data. Production clusters combine both: each shard is itself a replicated group of servers.