TopolearnProduction ClickHouse runs on clusters, and their design decides availability, cost and query fan-out.
ClickHouse scales in two independent directions.
Replication keeps identical copies of a table on several servers for availability and read capacity. Use a Replicated*MergeTree engine with a coordination path and replica name, usually via macros: ReplicatedMergeTree('/clickhouse/tables/{shard}/{database}/{table}', '{replica}'). ClickHouse Keeper (a Raft-based ZooKeeper replacement) stores the replication log. Each replica accepts inserts and fetches parts it's missing from the others; replication is asynchronous and multi-leader. Identical insert blocks are deduplicated, which makes retries safe.
Sharding splits rows across servers for capacity. Each shard holds local tables; a Distributed table, Distributed(cluster, db, local_table, sharding_key), holds no data itself. Queries fan out to one replica per shard, each shard aggregates its slice, and the initiator merges the partial results. Inserting through a Distributed table routes rows by the sharding key, but writing directly to local tables in big batches is often more reliable.
Design points:
- Pick a sharding key that keeps related rows together (e.g.
tenant_id) so many queries hit one shard, and soGROUP BYandJOINstay local. GLOBAL IN/GLOBAL JOINsend a subquery result to every shard once, instead of each shard re-running it.ON CLUSTER my_clusterruns DDL on every node; theReplicateddatabase engine replicates DDL automatically.
ClickHouse Cloud separates storage and compute: its SharedMergeTree engine keeps data in object storage, so you scale replicas without manual sharding.
Example
-- On every node (config defines cluster 'events_cluster' and macros {shard}, {replica})
CREATE TABLE events_local ON CLUSTER events_cluster
(
tenant UInt32,
ts DateTime,
kind LowCardinality(String)
)
ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/{database}/events_local', '{replica}')
PARTITION BY toYYYYMM(ts)
ORDER BY (tenant, kind, ts);
-- Query/insert entry point that fans out to all shards
CREATE TABLE events ON CLUSTER events_cluster AS events_local
ENGINE = Distributed(events_cluster, currentDatabase(), events_local, tenant);
-- Each shard aggregates its rows; the initiator merges partial results
SELECT kind, count() FROM events WHERE ts > now() - INTERVAL 1 DAY GROUP BY kind;Remember
- ReplicatedMergeTree + Keeper = copies; Distributed tables = shards
- Shard by a key that keeps related rows together, such as tenant
- Use ON CLUSTER or the Replicated database engine to keep DDL in sync
Watch out for
- Sharding by rand() and then joining or counting distinct per tenant, forcing cross-shard data movement.