Scaling ClickHouse Horizontally: Sharding, Distributed Tables, and Parallel Replicas

Horizontal scale is where most ClickHouse fleets accumulate their deepest operational debt. ClickHouse sharding gets introduced months before the workload justifies it, and once a sharding key is baked into a production Distributed table, unwinding it becomes a data migration project. This playbook is written for engineers already running ReplicatedMergeTree in production who need to choose the next topology move. It covers scale-up versus shard, Distributed engine mechanics, replica routing, parallel replicas as an alternative read-scaling axis, ClickHouse Keeper quorum sizing, and resharding. Every change below carries a blast radius and rollback note, because on a live cluster nothing is reversible by default.

Scale Up Before You Commit to ClickHouse Sharding

A single node absorbs far more load than most capacity plans assume. Query parallelism scales with cores through max_threads, scan throughput scales with NVMe bandwidth, and repeated lookups scale with page cache. Doubling vCPU and RAM is a maintenance window. Adding a shard permanently changes your data topology, backup surface, DDL propagation path, and on-call runbook.

Vertical moves are also trivially reversible: you resize back. That asymmetry alone should push the ClickHouse sharding decision as late as your capacity curve allows. Two hard limits genuinely force horizontal growth. The first is storage, where the working set no longer fits the largest instance or array you can procure. The second is write throughput, where one node cannot absorb the insert rate even after batching, asynchronous inserts, and merge tuning.

Why premature ClickHouse sharding is the classic failure mode

Premature ClickHouse sharding fails for a predictable reason: the sharding key is chosen before query patterns stabilise. Six months later the dominant access pattern filters on a column that is not the sharding key, so every query fans out to every shard and the coordinator merges results it should never have received. You pay distributed overhead on every request and collect none of the pruning benefit.

The second failure is JOIN topology. When two large tables are sharded on different keys, correlated queries fall back to GLOBAL JOIN, which materialises the right-hand side on the initiator and broadcasts it to every shard. That network cost grows with cluster width, converting a fast local hash join into a cluster-wide event.

Signals that genuinely justify ClickHouse sharding

These are the conditions under which ClickHouse sharding actually pays for itself, and all of them are measurable before you commit:

• Total bytes in system.parts on the busiest node trending toward disk capacity with no codec, TTL, or tiered-storage headroom left.
• Sustained insert backpressure showing as growing inserts_in_queue and merges_in_queue in system.replicas after batching is already tuned.
• CPU saturation during scans that persists after max_threads, PREWHERE, and skip-index work, on the largest instance type available to you.
• A natural, stable, high-cardinality partitioning dimension such as tenant or account that most queries already filter on.

Distributed Table Mechanics Behind ClickHouse Sharding

ClickHouse sharding is implemented by two objects: a local replicated table on every node, and a Distributed table that stores no data and fans queries out. Reads are parallelised across shards automatically, remote indexes are used, and aggregation is partially executed on each shard before intermediate states return to the initiator.

Declaring the cluster and the Distributed table

-- Local storage table, created on every replica of every shard.
CREATE TABLE analytics.events_local ON CLUSTER analytics_cluster
(
    event_date  Date,
    event_time  DateTime,
    tenant_id   UInt32,
    user_id     UInt64,
    event_type  LowCardinality(String),
    payload     String
)
ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/analytics/events_local', '{replica}')
PARTITION BY toYYYYMM(event_date)
ORDER BY (tenant_id, event_type, event_time)
SETTINGS
    index_granularity = 8192,
    min_bytes_for_wide_part = 10485760;

-- Fan-out table. The sharding key is explicit, integer-valued, and hashed.
CREATE TABLE analytics.events_distributed ON CLUSTER analytics_cluster
(
    event_date  Date,
    event_time  DateTime,
    tenant_id   UInt32,
    user_id     UInt64,
    event_type  LowCardinality(String),
    payload     String
)
ENGINE = Distributed('analytics_cluster', 'analytics', 'events_local', intHash64(tenant_id))
SETTINGS
    fsync_after_insert = 0,
    fsync_directories = 0,
    background_insert_batch = 1;

The ClickHouse sharding expression must be any expression over constants and columns that returns an integer. ClickHouse takes its remainder against the total shard weight to pick a destination. Wrap skewed columns in a hash function, as above, or one shard inherits your largest tenant. Sharding on tenant_id colocates each tenant, which is what makes local IN and JOIN possible and lets optimize_skip_unused_shards prune fan-out.

Shard weights, internal_replication, and the duplicate-write trap

Each shard carries a <weight> that defaults to 1 and controls its share of inserts proportionally. Each shard also carries <internal_replication>, which defaults to false. With the default, the Distributed table writes to every replica itself and no consistency check is performed, so replicas drift over time. Any cluster backed by a Replicated*MergeTree engine should set it to true so one healthy replica takes the write and native replication propagates it.

Blast radius: flipping internal_replication on a cluster that has been writing with it disabled does not repair divergence that already exists. New writes behave correctly; historical drift stays. Rollback: revert the config file and reload. No restart is needed, but schedule a comparison of row counts and part checksums across replicas before you declare the change complete.

distributed_product_mode and the real cost of GLOBAL JOIN

When a subquery inside IN or JOIN references a distributed table, distributed_product_mode decides what happens. The default is deny, which raises an exception rather than silently choosing a strategy. The alternatives are local, which rewrites the inner reference to the local table on each shard; global, which converts the query to GLOBAL IN or GLOBAL JOIN; and allow, which permits the distributed product and is almost never what you want at scale.

GLOBAL JOIN executes the right-hand side once on the initiator, stores it in a temporary table, and ships that table to every shard. The network and memory cost therefore scales with right-hand-side size multiplied by shard count, which is why a query that is comfortable on three shards can fall over on thirty.

Blast radius: setting distributed_product_mode = 'local' when data is not truly colocated by the sharding key produces silently incomplete results rather than an error. This is the sharpest edge in the entire ClickHouse sharding surface. Rollback: revert the session or profile value; it takes effect on the next query with no restart, but any downstream artefacts built from wrong results must be recomputed.

Replica Routing with ReplicatedMergeTree and load_balancing

ClickHouse sharding distributes data; replication distributes availability. Within each shard, replicas coordinate through ClickHouse Keeper, and the Distributed table selects one replica per shard for each read. The load_balancing setting governs that selection, and the wrong choice concentrates read load on a subset of nodes while the rest idle.

Replica avoidance is driven by errors_count and slowdowns_count in system.clusters, decayed by distributed_replica_error_half_life and capped by distributed_replica_error_cap. A replica that flaps will keep receiving traffic until its error count accumulates, so check estimated_recovery_time before blaming the balancer.

Blast radius: changing load_balancing in the default profile shifts read traffic cluster-wide on the next query, which can cold-start filesystem caches on previously idle replicas and cause a transient latency spike. Rollback: revert the profile value and reload; effective immediately, no restart.

Parallel Replicas: Scaling Reads Without ClickHouse Sharding

Parallel replicas attack a different axis from ClickHouse sharding. Instead of splitting data across shards, they split a single query across the replicas of one shard, using granules rather than shards as the unit of work. The node receiving the query becomes coordinator, collects announcements describing which parts each replica actually holds, assigns granule ranges accordingly, and merges the mergeable states that come back.

Two mechanisms make this survivable in production. Dynamic coordination lets replicas request additional work when they finish early instead of receiving one fixed batch, which bounds tail latency. Task assignment is hashed over part and granule set so repeat queries hit warm caches, and slower replicas have their pending tasks stolen by faster ones.

Version-pinning the feature and its maturity

The capability first shipped behind allow_experimental_parallel_reading_from_replicas and is now controlled by enable_parallel_replicas, which accepts 0 for off, 1 for enabled, and 2 to force usage and throw when it cannot be applied. Execution requires the analyzer, so enable_analyzer must be on. Documented limitations still apply: parallel replicas are disabled with FINAL, projections are not used alongside them, CTEs and JOINs can regress, and small queries lose to coordination overhead.

Maturity varies by build and by release, so pin it against your own binary rather than a blog post — including this one:

SELECT
    name,
    value,
    changed,
    tier
FROM system.settings
WHERE name LIKE '%parallel_replicas%'
ORDER BY name ASC;

Blast radius: enabling this in the default profile changes execution for every query on the cluster, including short interactive ones that will get slower. Scope it to a dedicated user profile and validate with EXPLAIN PIPELINE before widening. Rollback: SET enable_parallel_replicas = 0, effective on the next query, no restart and no data movement. That free rollback is exactly why parallel replicas deserve a trial before ClickHouse sharding.

ClickHouse Keeper Quorum Sizing and Placement

Keeper is a Raft ensemble, so a majority must acknowledge every write: three nodes tolerate one failure, five tolerate two, and even-sized ensembles buy nothing. Keeper writes sit on the critical path for part commits, mutations, and distributed DDL, which means quorum latency is insert latency for every ReplicatedMergeTree table you own.

Multi-shard ClickHouse sharding raises Keeper load, but the driver is the number of replicated tables multiplied by parts created per second, not the shard count on its own. An ingest pipeline producing many small parts will stress a three-node ensemble long before a wide cluster with disciplined batching does. Growing from three to five nodes increases fault tolerance and increases commit latency at the same time; make that trade knowingly.

Placement rules for multi-shard ClickHouse sharding

Spread three Keeper nodes across three independent failure domains. Two availability zones cannot produce a safe quorum, because losing the zone holding two nodes loses the majority. If you genuinely have only two zones, place a third lightweight Keeper — with no ClickHouse server attached — in a third location. Give Keeper a dedicated disk: its log fsync competes directly with merge I/O, and co-locating it with your heaviest merge workload is a reliable way to manufacture latency.

Blast radius: a botched ensemble change loses quorum, and every replicated table across every shard flips to read-only with is_readonly = 1 in system.replicas. Inserts fail cluster-wide within seconds. Rollback: restore the previous raft_configuration and restart Keeper nodes one at a time, never in parallel, verifying quorum is re-established before touching the next node.

Resharding Strategies and Their Operational Cost

ClickHouse has no online resharding primitive. Resharding is the bill for a ClickHouse sharding decision made too early, and every strategy below is a data-movement project with a maintenance budget attached.

Weighted drain

Add new shards, then lower or zero the weights of the old ones so new writes land on new capacity while historical data ages out under TTL. This is the lowest-risk option and only works for time-series data with finite retention. Blast radius: query fan-out is uneven for the length of the retention window, so per-shard resource use diverges. Rollback: restore original weights via config reload; no data moved, so the rollback is genuinely clean.

Dual-write and backfill

Write to old and new clusters concurrently from the ingestion layer, backfill history with INSERT INTO ... SELECT through remote(), then cut reads over once row counts reconcile. Blast radius: doubled write cost and two candidate sources of truth until cutover. Rollback: point reads back at the old cluster, which remains authoritative for as long as you keep dual-writing — do not stop dual-writing on cutover day.

Partition-level move

Valid only when the sharding key aligns with the partition key. Fetch and attach partitions on the target shard, then drop the source copy in a separate window.

ALTER TABLE analytics.events_local ON CLUSTER analytics_cluster
    FETCH PARTITION '202607'
    FROM '/clickhouse/tables/1/analytics/events_local';

ALTER TABLE analytics.events_local
    ATTACH PARTITION '202607';

Blast radius: the attach is near-instant but the fetch is a full network copy that competes with production reads. Rollback: DETACH PARTITION on the target; the source copy is untouched until you explicitly drop it, which is why dropping the source must never share a maintenance window with the move.

Verifying ClickHouse Sharding Health from System Tables

Three system tables answer nearly every ClickHouse sharding topology question. Run these before and after any change, and keep the before-output.

-- Topology as the server actually resolved it, plus replica error state.
SELECT
    cluster,
    shard_num,
    shard_weight,
    replica_num,
    host_name,
    is_local,
    errors_count,
    slowdowns_count,
    estimated_recovery_time
FROM system.clusters
WHERE cluster = 'analytics_cluster'
ORDER BY
    shard_num ASC,
    replica_num ASC;

-- Replication health for every replicated table on this node.
SELECT
    database,
    table,
    is_readonly,
    is_session_expired,
    absolute_delay,
    queue_size,
    inserts_in_queue,
    merges_in_queue,
    total_replicas,
    active_replicas
FROM system.replicas
WHERE database = 'analytics'
ORDER BY absolute_delay DESC;

-- Backlog of asynchronous Distributed INSERTs waiting to reach shards.
SELECT
    database,
    table,
    is_blocked,
    error_count,
    data_files,
    data_compressed_bytes,
    broken_data_files,
    last_exception,
    last_exception_time
FROM system.distribution_queue
ORDER BY data_compressed_bytes DESC;

Read them together. is_readonly = 1 means the Keeper session is gone, not that the disk is full. A single replica whose absolute_delay climbs while others hold steady should be pulled from routing rather than debugged in the request path. In system.distribution_queue, monotonically rising data_files means a shard is unreachable or rejecting writes, and any non-zero broken_data_files indicates data already relegated to the broken directory that will never be sent without manual intervention.

Alert thresholds must come from your own baseline. Any figure quoted in an article is illustrative only; the numbers worth paging on are your steady-state values measured over a full weekly cycle.

Blast Radius and Rollback Cheat Sheet

Treat this as the pre-flight checklist for any ClickHouse sharding operation. Capture the current state before you touch anything.

Official references worth keeping open while you plan: the Distributed table engine documentation and the system.clusters reference.

Test in Staging First, Then Re-Check Your DR Posture

Every ClickHouse sharding command, config delta, and setting above should be rehearsed in a staging cluster that mirrors production shard count, replica count, and Keeper topology. A two-node laboratory will not reproduce quorum behaviour, fan-out cost, or GLOBAL JOIN pressure. Rehearse the rollback as carefully as the change itself, and time both.

ClickHouse sharding also rewrites your disaster-recovery maths. Each new shard is an independent failure domain holding a slice of the dataset that no replica outside that shard can cover. Before you widen the cluster, confirm that backups run per shard and restore as a consistent set, that Keeper state is captured alongside table data, and that you have a tested procedure for rebuilding one lost shard while the remainder keeps serving. Replication is not a backup, and a wider cluster only widens the surface.

Planning a shard split, a Keeper redesign, or a parallel-replicas rollout? ChistaDATA’s ClickHouse Managed Services team runs these migrations on live production clusters with rehearsed rollback plans and measured cutovers, and our ClickHouse DBA engineers will review your topology before you commit to a sharding key you cannot change. Talk to ChistaDATA Managed Services about a ClickHouse sharding and scalability review of your cluster.