12. Distributed ClickHouse

One ClickHouse server goes a very long way — many production deployments are a single machine with a lot of disk, handling hundreds of billions of rows. But eventually you need more storage than one machine holds, more throughput than its cores provide, or redundancy so a failure is not an outage.

ClickHouse separates those two needs cleanly, and understanding the separation is most of this chapter.

Two independent axes

Replication — the same data on several servers. For availability and read capacity. Implemented by ReplicatedMergeTree.

Sharding — different data on different servers. For storage and write capacity. Implemented by the Distributed engine.

They are orthogonal. You can have replication without sharding (common, and usually where you should start), sharding without replication (fast, fragile), or both.

   SHARD 1              SHARD 2              SHARD 3
   ┌──────────┐         ┌──────────┐         ┌──────────┐
   │ replica1 │◀──────▶ │ replica1 │◀──────▶ │ replica1 │
   │ replica2 │  same   │ replica2 │  same   │ replica2 │
   └──────────┘  data   └──────────┘  data   └──────────┘
        │  within a shard      │                   │
        └──────────────────────┴───────────────────┘
                    different data across shards

Replication — ReplicatedMergeTree

CREATE TABLE page_views ON CLUSTER my_cluster (
    ts DateTime, tenant_id UInt32, country LowCardinality(String), url String
)
ENGINE = ReplicatedMergeTree(
    '/clickhouse/tables/{shard}/page_views',    -- path in Keeper
    '{replica}'                                  -- this replica's name
)
PARTITION BY toYYYYMM(ts)
ORDER BY (tenant_id, country, ts);

{shard} and {replica} are macros defined per server in its config, so the same DDL runs everywhere.

How it works

Replication is per table, multi-master, and asynchronous.

1. You insert into any replica. There is no primary.
2. That replica writes the part locally and registers it in the replication log in Keeper.
3. Other replicas see the log entry and fetch the part over HTTP from the replica that has it.
4. Merges are also coordinated through Keeper: one replica is assigned the merge, performs it, and the others fetch the merged part rather than repeating the work.

That last point is important and often missed. Replicas do not each independently merge — the merge is done once and the result is copied. This saves an enormous amount of CPU and guarantees the replicas’ parts are byte-identical.

Keeper

Coordination requires consensus, provided by ClickHouse Keeper — a C++ reimplementation of ZooKeeper’s protocol, built into ClickHouse.

Keeper holds: the list of replicas, the replication log, part metadata, merge assignments, insert deduplication hashes, and leader election.

Keeper is a hard dependency for replicated tables. If Keeper is unavailable:

• Inserts into replicated tables fail (they cannot register the new part).
• Merges stop.
• Reads keep working — this is the saving grace, and it means a Keeper outage degrades you to read-only rather than down.

Run three or five Keeper nodes — an odd number, for majority quorum. It can run inside clickhouse-server processes or as separate clickhouse-keeper processes; separate is better in production, because Keeper wants low latency and predictable resources and should not compete with query execution.

If you are running ZooKeeper for historical reasons, Keeper is a drop-in replacement that uses substantially less memory and is simpler to operate. Migrate. The ZooKeeper guide’s chapter 12 covers what Keeper actually reimplements — the client protocol, over Raft instead of ZAB — and why the absence of a JVM matters so much for a coordination service.

SELECT * FROM system.zookeeper WHERE path = '/clickhouse/tables';
SELECT name, is_leader, total_replicas, active_replicas,
       absolute_delay, queue_size, inserts_in_queue, merges_in_queue
FROM system.replicas;

system.replicas is the health check for this chapter. absolute_delay is replication lag in seconds; queue_size is how much work a replica has yet to do. Both climbing means a replica is falling behind.

Insert quorum

By default an insert is acknowledged as soon as the local replica has it — replication happens afterwards. If that server dies immediately, the data can be lost.

SET insert_quorum = 2;
SET insert_quorum_timeout = 60000;
SET select_sequential_consistency = 1;

Now the insert waits for two replicas. This is the same trade as MongoDB’s w: "majority" or Kafka’s acks=all: correctness for latency.

select_sequential_consistency makes reads wait for the replica to have caught up to the last quorum-committed insert, which gives read-your-writes at the cost of blocking.

Most analytics workloads leave quorum off and accept that a server loss may cost the last few seconds of ingestion — because the data usually came from Kafka and can be replayed. Decide deliberately rather than by default.

Sharding — the Distributed engine

A Distributed table stores no data. It is a view over the same table on many servers — a router, like MongoDB’s mongos.

config.xml

<remote_servers>
  <my_cluster>
    <shard>
      <internal_replication>true</internal_replication>
      <replica><host>ch-1a</host><port>9000</port></replica>
      <replica><host>ch-1b</host><port>9000</port></replica>
    </shard>
    <shard>
      <internal_replication>true</internal_replication>
      <replica><host>ch-2a</host><port>9000</port></replica>
      <replica><host>ch-2b</host><port>9000</port></replica>
    </shard>
  </my_cluster>
</remote_servers>

CREATE TABLE page_views_dist AS page_views
ENGINE = Distributed(my_cluster, default, page_views, cityHash64(tenant_id));

Four arguments: the cluster, the database, the local table name, and the sharding key.

internal_replication must be true

With internal_replication = true, the Distributed table sends an insert to one replica per shard and lets ReplicatedMergeTree handle copying it. With false, it writes to every replica itself — which duplicates the work and, worse, can produce divergent replicas if one write fails.

If your tables are ReplicatedMergeTree, set it to true. This is a footgun that has bitten many deployments.

Reading

A SELECT on a Distributed table:

1. The initiator node rewrites the query for each shard.
2. Each shard runs it locally, in parallel, on its own data.
3. Shards return partial results — partial aggregation states where possible, not raw rows.
4. The initiator merges them and produces the final answer.

The key efficiency: for SELECT country, count() … GROUP BY country, each shard sends back its own small aggregation, not its rows. Network transfer is proportional to the result, not the data.

This breaks down for queries that cannot be pre-aggregated per shard — a large ORDER BY without a LIMIT, uniqExact across shards, some window functions. Those pull data to the initiator and the initiator becomes the bottleneck.

distributed_group_by_no_merge and optimize_distributed_group_by_sharding_key let you skip the merge step when the GROUP BY key aligns with the sharding key — a large win when it applies.

Writing

Two options, and the second is usually right.

Write to the Distributed table. Convenient. Rows are routed by the sharding key, and by default buffered on disk and forwarded asynchronously (distributed_foreground_insert = 0, historically named insert_distributed_sync). That async buffering is a durability gap: rows sitting in a local spool are lost if the node dies.

Write directly to the local tables, with your ingestion layer choosing the shard. More work, more control, no spool. This is what most large deployments do — the Kafka consumer on each node reads its own partitions and writes locally, so sharding follows Kafka partitioning naturally.

Choosing a sharding key

Same considerations as chapter 10 of the MongoDB guide, and the same failure modes:

• Even distribution. cityHash64(user_id) or rand().
• Query locality. If most queries filter by tenant_id, sharding on cityHash64(tenant_id) means those queries hit one shard. A shard-local query is far cheaper than a fan-out.
• Join co-location. Two tables sharded by the same key can be joined locally on each shard, with no GLOBAL JOIN (chapter 10).
• Skew. Hashing a tenant id when one tenant is 60% of your data puts 60% of the data on one shard. Check before committing.

rand() distributes perfectly and destroys all locality. It is right when queries always scan everything anyway, and wrong otherwise.

Changing the sharding key means redistributing the data. There is no automatic rebalancer — you create a new distributed table and copy, or use clickhouse-copier-style tooling. Choose carefully.

Distributed DDL

ON CLUSTER runs DDL everywhere:

CREATE TABLE … ON CLUSTER my_cluster (…) ENGINE = ReplicatedMergeTree(…);
ALTER TABLE page_views ON CLUSTER my_cluster ADD COLUMN referrer String;
DROP TABLE page_views_dist ON CLUSTER my_cluster SYNC;

The statement goes into a Keeper queue and each node executes it. Watch for stragglers:

SELECT * FROM system.distributed_ddl_queue WHERE status != 'Finished';

A node that was down when the DDL ran will execute it on startup — which is good, and means a cluster can be briefly inconsistent in schema. Check the queue after any cluster-wide change.

The shared-storage model

Worth knowing because it is where ClickHouse deployment is heading.

The classic architecture above is shared-nothing: each node owns its disk, and replication copies data between them.

Newer deployments — ClickHouse Cloud, and the open-source SharedMergeTree direction — separate storage from compute: all data lives in object storage (S3), and stateless compute nodes read from it, with a local disk cache for hot data.

The consequences are appealing: replicas are free (they share the same storage), scaling compute is instant, and storage grows without touching the cluster. The costs are object-storage latency, which the cache mostly hides, and dependence on a specific implementation.

You can get partway there in open-source ClickHouse today with an s3 disk and tiered TTL (chapter 5), which is a genuinely useful middle ground: hot data local, cold data in S3, one table.

Do you need a cluster?

Ask honestly before building one.

One large server handles more than people expect. 64 cores, 512 GB RAM and 50 TB of NVMe will hold hundreds of billions of compressed rows and scan billions of rows per second. Many companies run their entire analytics on one machine plus a replica.

Add replication first. It gives you availability and read capacity, and it is much simpler than sharding — Keeper plus ReplicatedMergeTree, no distributed query planning, no sharding key.

Add sharding when one machine genuinely cannot hold or ingest the data. Not before. Sharding brings the distributed query costs, the sharding key decision, GLOBAL JOIN subtleties, and a materially harder operational story.

The escalation order: bigger machine → better schema (chapters 2, 5, 11) → replication → sharding.

What you would type

-- cluster topology
SELECT cluster, shard_num, replica_num, host_name, is_local
FROM system.clusters ORDER BY cluster, shard_num, replica_num;

-- replication health — the key check
SELECT database, table, is_readonly, absolute_delay, queue_size,
       inserts_in_queue, merges_in_queue, last_queue_update_exception
FROM system.replicas WHERE absolute_delay > 10 OR queue_size > 100;

-- what a replica is waiting on
SELECT * FROM system.replication_queue WHERE num_tries > 3;

-- pending distributed DDL
SELECT * FROM system.distributed_ddl_queue WHERE status != 'Finished';

-- run something on every node
SELECT hostName(), count() FROM clusterAllReplicas(my_cluster, system.parts)
WHERE active GROUP BY hostName();

That last pattern — clusterAllReplicas over a system table — is the fastest way to check a condition across the whole cluster from one connection.

What goes wrong

internal_replication = false with replicated tables. Duplicate work and divergent replicas.

Keeper down. All inserts to replicated tables fail; reads continue. Keeper needs its own monitoring, its own disk, and enough memory.

Keeper co-located with heavy query nodes. It gets starved of CPU and latency spikes propagate into every insert. Separate it.

A replica stuck in read-only. Usually lost its Keeper session or found data inconsistent. SYSTEM RESTORE REPLICA and SYSTEM SYNC REPLICA are the tools; check is_readonly and last_queue_update_exception first.

Replication queue growing. A replica cannot keep up — often a slow disk or an enormous merge. Watch queue_size.

A skewed sharding key. One shard holds most of the data and answers most of the queries.

Distributed inserts with async spool losing buffered rows when a node dies.

Sharding when replication would have done. Complexity with no benefit.

What to take from this chapter

• Replication and sharding are separate. ReplicatedMergeTree for the first, Distributed for the second. Start with replication.
• Replication is multi-master and asynchronous, coordinated by Keeper, and merges happen once and are fetched rather than repeated per replica.
• Keeper down = no inserts, no merges, reads still fine. Run 3 or 5 nodes, separate from query servers.
• system.replicas — absolute_delay and queue_size — is the health check.
• insert_quorum is ClickHouse’s w: "majority"; most analytics workloads leave it off deliberately because the source data is replayable.
• internal_replication = true whenever your tables are replicated.
• A Distributed SELECT sends partial aggregation states back, not rows — so network cost is proportional to the result.
• Choose a sharding key for even distribution and query locality; changing it later means copying everything.
• Escalate in order: bigger machine → better schema → replication → sharding. One server goes remarkably far.

Next: keeping it running.