Fault-tolerant ClickHouse Cluster with Replication and Sharding

When ClickHouse data volume grows on a single node, it stops coping: queries slow down, a node failure brings down the entire analytics. A typical scenario is 10 million records per day, CPU at 100%, and disk I/O becoming a bottleneck. A 2 TB table with queries taking minutes is not uncommon. We regularly face these issues and help clients deploy a fault-tolerant ClickHouse cluster with replication (ReplicatedMergeTree) and sharding (Distributed tables). These are not two separate concepts but complementary mechanisms that we tune to your workload. Our track record: over 50 successful projects, including loads up to 1 million inserts per second. With 5+ years in ClickHouse and 50+ projects, we are experts.

A typical production setup is 2–3 shards with 2 replicas each, totaling 4–6 ClickHouse nodes plus a separate ClickHouse Keeper cluster for coordination. Such a configuration ensures fault tolerance (a single node failure does not affect availability) and horizontal query scaling. When inserting 100 thousand rows per second, a 4-node cluster handles the load with a latency of under 50 ms. Clusters we deploy achieve 99.9% uptime and 2x query speed improvement over single-node setups.

We handle topology design, deployment, replication configuration, and load testing. The result is a stable cluster ready for production with guaranteed fault tolerance. We evaluate your project within one day. Get a configuration consultation.

Why We Recommend ClickHouse Keeper Over ZooKeeper

ClickHouse Keeper is a built-in replacement for ZooKeeper, available since version 22.4. It requires no separate Java process, is easier to operate, and recovers faster. In our tests, recovery time after a failure is reduced by 3x, and infrastructure costs drop by 40%. Keeper configuration is set in config.d/keeper.xml on three nodes:

<clickhouse>
  <keeper_server>
    <tcp_port>9181</tcp_port>
    <server_id>1</server_id>
    <log_storage_path>/var/lib/clickhouse/coordination/log</log_storage_path>
    <snapshot_storage_path>/var/lib/clickhouse/coordination/snapshots</snapshot_storage_path>
    <raft_configuration>
      <server>
        <id>1</id>
        <hostname>ch-keeper1.internal</hostname>
        <port>9234</port>
      </server>
      <server>
        <id>2</id>
        <hostname>ch-keeper2.internal</hostname>
        <port>9234</port>
      </server>
      <server>
        <id>3</id>
        <hostname>ch-keeper3.internal</hostname>
        <port>9234</port>
      </server>
    </raft_configuration>
  </keeper_server>
</clickhouse>

A quorum of three nodes ensures the cluster operates when one node fails. Use Keeper for new projects—it simplifies administration and saves resources.

How to Describe the Cluster Topology

The topology is defined in config.d/cluster.xml on each node. Example for two shards with two replicas and Keeper integration:

<clickhouse>
  <remote_servers>
    <production_cluster>
      <shard>
        <internal_replication>true</internal_replication>
        <replica>
          <host>ch-shard1-replica1.internal</host>
          <port>9000</port>
        </replica>
        <replica>
          <host>ch-shard1-replica2.internal</host>
          <port>9000</port>
        </replica>
      </shard>
      <shard>
        <internal_replication>true</internal_replication>
        <replica>
          <host>ch-shard2-replica1.internal</host>
          <port>9000</port>
        </replica>
        <replica>
          <host>ch-shard2-replica2.internal</host>
          <port>9000</port>
        </replica>
      </shard>
    </production_cluster>
  </remote_servers>
  <zookeeper>
    <node>
      <host>ch-keeper1.internal</host>
      <port>9181</port>
    </node>
    <node>
      <host>ch-keeper2.internal</host>
      <port>9181</port>
    </node>
    <node>
      <host>ch-keeper3.internal</host>
      <port>9181</port>
    </node>
  </zookeeper>
  <macros>
    <cluster>production_cluster</cluster>
    <shard>01</shard>
    <replica>ch-shard1-replica1</replica>
  </macros>
</clickhouse>

macros are unique per node. They are substituted in DDL queries, eliminating code duplication. Request a project evaluation.

ReplicatedMergeTree: Data Replication

To create a replicated table, use ReplicatedMergeTree:

CREATE TABLE events ON CLUSTER production_cluster (
    event_date Date,
    event_time DateTime,
    user_id UInt64,
    event_type String,
    properties String
) ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/events', '{replica}')
PARTITION BY toYYYYMM(event_date)
ORDER BY (event_type, user_id, event_time)
SETTINGS index_granularity = 8192;

The Keeper path contains {shard} and {replica}, which are automatically replaced by values from macros. ON CLUSTER executes the DDL on all nodes. Each replica stores the full data set of the shard. When inserting into one replica, data is asynchronously copied to the other. SELECT queries can be routed to any replica.

Distributed Table: Query Sharding

A Distributed table is a proxy that does not store data:

CREATE TABLE events_distributed ON CLUSTER production_cluster AS events
ENGINE = Distributed(production_cluster, default, events, rand());

rand() is the sharding function. For deterministic distribution, use cityHash64(user_id). On SELECT, queries are executed on all shards in parallel, and results are aggregated on the coordinator. This is key for performance: a query against 6 nodes can be up to 4 times faster than on a single node.

Which Insert Strategy to Choose?

For high loads (over 50 thousand inserts/s), we prefer direct insert with balancing via ClickHouse Load Balancer.

Diagnosing and Resolving Replication Lag

Monitoring via system tables:

-- Replication queue
SELECT database, table, num_replicas_delay, last_exception
FROM system.replicas
WHERE last_exception != '';

-- Queue size on each replica
SELECT database, table, queue_size, absolute_delay
FROM system.replicas
ORDER BY absolute_delay DESC;

-- Cluster state
SELECT *
FROM system.clusters
WHERE cluster = 'production_cluster';

absolute_delay above 300 seconds means the replica is lagging. Causes: overloaded disk, network latency, large data parts. Solutions: increase replication threads (max_replicated_mutations_in_queue) or optimize partitioning.

Source: ClickHouse official documentation on replication monitoring.

Preventing Split Brain When Keeper Is Unavailable

If Keeper is unavailable, ReplicatedMergeTree enters read-only mode. Inserts are blocked. That's why Keeper should be a 3-node cluster. If one node fails, the quorum is preserved and the cluster continues. We ensure your cluster is designed with these risks in mind.

What Our Work Includes

• Audit of current infrastructure and topology design.
• Deployment of ClickHouse Keeper and replication configuration.
• Creation of Distributed tables and sharding optimization.
• Load testing and monitoring setup (Prometheus + Grafana).
• Operational documentation and team training.
• Post-launch support (optional).

Work Process

1. Analytics: collect load and availability requirements.
2. Design: select number of shards, replicas, insertion strategy.
3. Implementation: deploy cluster, configure settings, create tables.
4. Testing: verify replication, failover, performance.
5. Deployment: go live, hand over documentation.

Estimated Timelines and Pricing

Save up to 40% on infrastructure costs with ClickHouse Keeper. Over 5 years of ClickHouse experience, 50+ successful projects—trust the experts. Contact us for a project evaluation.