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));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.
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.
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.