ClickHouse Cluster Sharding and Distributed Table Design

ClickHouse is a columnar, open-source database built for one job: answering analytical queries over huge volumes of raw data, fast. It is not a general-purpose distributed system, and most of the decisions that go wrong in production clusters come from treating it like one. Sharding, in particular, is not a scaling switch you flip when a cluster feels slow. It is a sequence of decisions, each one constraining the next, and each mistake compounding the moment real data lands on disk.

Columnar storage and vectorized, parallelized execution make ClickHouse fast. Columnar storage means a query only reads the columns it actually needs, not the whole row. Execution is vectorized within a node and parallelized across the cluster, so CPU cores and machines both get used at once instead of one after another. The architecture is shared-nothing: nodes don't share disk or memory, and each shard is a fully independent unit with its own storage and its own copy of whatever data it holds. That independence is why sharding decisions are so hard to undo later.

Given all that, ClickHouse's own design preference runs against the industry's default instinct. Most distributed databases assume horizontal scale from day one. ClickHouse assumes the opposite: scale the box first. Add CPU, add RAM, add faster disks, before adding a second node. That preference is not a limitation of the engine; it's a reflection of how much a single well-tuned node can actually do.

The two conditions that justify adding shards

There are exactly two legitimate reasons to shard a ClickHouse cluster. Either the data no longer fits on one server, or one server can't process the query workload fast enough. That's the whole list. Anything else, including "the cluster feels like it should be bigger" or "sharding is what serious deployments do," is premature complexity dressed up as engineering judgment.

Treat these as hard thresholds, not vibes. A team that shards because a dashboard query takes four seconds instead of one, without checking whether that's a hardware ceiling or a schema problem, has usually just added operational risk without solving anything.

Shards and replicas are not the same axis, and confusing this distinction leads directly to the wrong topology. Shards are disjoint subsets of data, they exist to increase write throughput and total storage capacity. Replicas are copies of a shard, they exist to increase read capacity and to keep the cluster alive when a node dies. A team that needs more read throughput and adds shards instead of replicas ends up with more operational surface area and no actual performance gain on the query pattern they were trying to fix.

Part of why the bar for sharding sits higher than people expect comes down to how ClickHouse handles writes and storage under the hood. Its async batch insert model and background merge process let a single, properly sized node hold billions of rows while still answering queries in near-real time. The instinct to shard early usually comes from experience with row-oriented, non-columnar systems, where that kind of scale on one box simply isn't possible. In ClickHouse, it's the starting assumption.

Three table patterns and where each belongs in your schema

Every distributed ClickHouse deployment, no matter how it's dressed up, resolves into three fundamental table patterns. Knowing which one belongs where is most of what separates a clean schema from one that quietly falls over under join load.

All-sharded tables split data into disjoint pieces across nodes, with each node holding only its slice. This pattern appears in system tables and in genuinely large fact data. All-replicated tables put a full copy of the data on every node, which is the correct pattern for small dimension tables, the kind that need to be join-local everywhere so a query never has to reach across the network for a lookup. Sharded-plus-replicated is the standard production pattern for large fact tables: data gets split across shards for capacity, and each shard carries one or more replicas so a disk failure doesn't take data offline.

Replication in ClickHouse is a table-level property. That matters more than it sounds like it should. A single node can hold a replicated version of one table and a plain, non-replicated version of another, simultaneously, with no conflict. Nothing about the server itself commits it to one pattern.

The rule that falls out of this is simple, even if it feels counterintuitive at first: dimension tables are almost always stored as a full copy on every node, because doing so eliminates cross-shard joins for lookups. The fact table is the one that gets sharded. Concretely, every distributed deployment needs three objects: a ReplicatedMergeTree local table that actually stores the data on each node, a Distributed table that holds zero data and exists purely as a routing layer, and, where needed, a replicated dimension table configured to live in full on every node.

What goes in remote_servers and why each parameter matters

Cluster topology gets declared under <remote_servers>, either directly in config.xml or in a drop-in file such as config.d/clusters.xml. Every node in the cluster needs the identical <remote_servers> block. What differs from node to node is a small configuration block, which is what lets one block of markup describe a cluster from four different vantage points.

Inside each shard definition, two parameters carry more weight than their brevity suggests. The weight parameter controls how data gets distributed: ClickHouse sums all shard weights and gives each shard a proportional share of new data. Equal weight across shards is the default, and it's the right call unless the hardware itself is asymmetric, say, one node in the cluster has meaningfully more disk than the others, in which case an unequal weight lets that node absorb a larger share. The internal_replication flag, when set to true, tells ClickHouse to write to just one replica per shard rather than duplicating the write across every replica in that shard. For any setup built on ReplicatedMergeTree, that's the safe default, since replication is already handled at the table engine level and duplicating writes on top of it just wastes bandwidth.

A concrete reference topology helps make this less abstract. A guide from oneuptime lays out a two-shard, two-replica-per-shard cluster across four nodes (ch-node-01 through ch-node-04), which is about as small as a "real" production topology gets while still exercising every part of the configuration.

The mechanism that ties a shared remote_servers block to per-node identity is the {shard} and {replica} macros. Set per node and referenced inside table DDL, they're what let a single ON CLUSTER statement create correctly named, correctly pathed tables across every node in the cluster, without an operator hardcoding a different path by hand on each one.

Choosing the sharding key: the decision that cannot be undone without re-inserting all data

Of every decision in this list, the sharding key is the one with no cheap way back. Once data has been loaded against a given key, changing it means re-inserting the entire dataset. There's no migration path, no schema-altering shortcut, no shortcut. A synthetic key derived from a stable business identifier may make more sense than reaching for whatever column is closest at hand, and that single fact should shape how much time gets spent on this decision upfront.

A good sharding key satisfies three properties at once, and it's rare for the obvious choice to satisfy all three without some thought. It needs to spread rows evenly across shards. It needs to keep rows that get queried together on the same shard, so a query doesn't have to scatter out to every shard and gather the results back (one of the most costly patterns in a distributed query). And it needs to be cheap to compute at insert time, since that computation happens on every single row written.

intHash64(user_id) is the standard choice for user-centric analytics, and for good reason: it's a fast hash function, it produces a uniform spread, and it colocates every row belonging to a given user onto one shard. intHash64(tenant_id) A query filtering on WHERE tenant_id = 42 only ever touches one shard, a strong property for single-tenant query patterns. A composite key like cityHash64(tenant_id, user_id) distributes at the combined tenant-and-user grain, which keeps per-user data together while avoiding the hot-shard problem that a pure tenant-id key would create if one tenant happens to be enormous. And rand() gives the best possible distribution with zero colocation, which is only the correct choice when every query is expected to scatter across all shards anyway and the sole goal is even disk usage.

Each anti-pattern looks reasonable until the data actually lands. A low-cardinality column, something like a status field with roughly ten possible values, produces a classic hot-shard problem: most shards end up empty or nearly so, because there simply aren't enough distinct values to spread across however many shards the cluster has. A key based purely on time creates a permanent write hot spot, since every insert for the current period lands on the same shard by definition. And columns with naturally skewed distributions cluster most rows into a handful of values no matter how many distinct values technically exist, which defeats the purpose of hashing.

Distributed table engine routing and the async insert tradeoff

The Distributed table engine stores no data of its own. It's a routing proxy, full stop. On a SELECT, it fans the query out as subqueries to every shard, and the initiator node, whichever node received the original query, merges the partial results back together. On an INSERT, it routes each row to whichever shard the sharding key expression says it belongs on.

In practice, that means every distributed schema follows the same two-step DDL pattern. The local table gets created first:

CREATE TABLE events_local ON CLUSTER my_cluster (...)
ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/events', '{replica}')
PARTITION BY toYYYYMM(event_time)
ORDER BY (user_id, event_time);

The Distributed table sits on top of it:

CREATE TABLE events ON CLUSTER my_cluster (...)
ENGINE = Distributed('my_cluster', 'default', 'events_local', intHash64(user_id));

Underneath, ClickHouse pushes the subquery down to the local table on each shard, runs it there, and merges partials at the initiator. The query planner will not silently compensate for a badly chosen sharding key or a poorly ordered ORDER BY clause. The schema has to be right, because the engine isn't going to fix it at query time.

Then there's the tradeoff that catches people who assume ClickHouse behaves like a strongly consistent system by default: async inserts. By default, ClickHouse buffers incoming inserts and sends them to shards asynchronously, which is a deliberate throughput optimization. The cost is that rows aren't immediately visible across all shards, and they aren't immediately durable on the remote shard the moment the client gets an acknowledgment. For most analytical workloads that's an acceptable trade. For anything that assumes read-after-write consistency across the cluster, it's a trap.

Distributed JOINs, GLOBAL IN, and the dangers of join design in distributed ClickHouse

Joins are where distributed ClickHouse becomes genuinely dangerous, because the default failure mode isn't slowness, it's silently wrong answers. When the right-hand side of a join is itself a distributed table, the default behavior has each shard execute the right-side subquery independently against only its own local slice of that data. If the right-side table is sharded, each shard only ever sees its own fragment of it. The join still runs, the query still returns rows, and those rows are simply incomplete or wrong, without necessarily any indication that something has gone awry.

The fix is the GLOBAL keyword, used as GLOBAL IN or GLOBAL JOIN. It changes the execution plan: the right-side query runs exactly once on the initiator node, the full result set gets collected into a temporary table, and that temporary table is broadcast out to every shard before the join executes locally. Correct results, at the cost of a network broadcast that has to move that entire result set to every node in the cluster.

That cost is also the limitation. GLOBAL JOIN only works when the right-side result set is small enough to broadcast without blowing out memory or saturating the network, and ClickHouse's own documentation warns against using GLOBAL IN against large datasets specifically because the network bandwidth used for that broadcast isn't configurable. There's no throttle to reach for if the result set turns out bigger than expected.

As of September 2026, there's an open, unresolved bug: SAMPLE combined with GLOBAL RIGHT or GLOBAL FULL JOIN over a sharded Distributed table with two or more shards can silently produce incorrect results due to wrong row sampling, leading to quiet row loss, or in other cases refuse outright with a SAMPLING_NOT_SUPPORTED error. Whether the issue is reachable without the GLOBAL keyword when distributed_product_mode = 'global' is set has not been confirmed. GitHub tracks the issue as #120838, and it remains open against current master. Anyone combining SAMPLE with outer joins on a sharded cluster should treat that combination as unverified until the issue closes.