← Back to the section

ClickHouse forgives a lot — a single server handles billions of rows — but it has its own set of operational quirks that look nothing like PostgreSQL or Kafka. Let's go over what you set up before production and what you watch after.

an insert creates a part on one replica, the others fetch it via the Keeper log replica A 202406_1 202406_2 replica B 202406_1 202406_2 Keeper replication log 202406_3INSERT: a new part GET_PART 202406_3 one entry in the queue the part travels directly, bypassing Keeper 202406_3queue is empty

You can write to any replica: the insert stores the data in a new part and appends a "fetch this part" entry to the Keeper log. The other replicas keep their own queue over that log and download parts from each other directly — only metadata goes through Keeper. The length of the not-yet-applied queue is exactly the replica's lag.

Replication: why it exists and how it works

If ClickHouse runs as a single instance and the server goes down, the data becomes unavailable. Replication helps by duplicating that data across several machines.

In ClickHouse, replication happens at the table level, not the server level. Only tables whose engine belongs to the Replicated*MergeTree family are replicated:

CREATE TABLE order_events ON CLUSTER main
(
    event_time   DateTime,
    event_type   LowCardinality(String),
    order_id     UUID,
    amount       Decimal(18, 2)
)
ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/order_events', '{replica}')
PARTITION BY toYYYYMM(event_time)
ORDER BY (event_type, event_time);

The replication coordinator is ClickHouse Keeper (the built-in replacement for ZooKeeper; for new installations, use Keeper only). Keeper stores metadata: which data parts exist, the replication queue, which replica schedules merges. The data itself travels between replicas directly.

Important properties:

  • Replication is asynchronous and multi-master — you can write to any replica and the others will catch up. A read from another replica immediately after a write may not see the data — that's fine for analytical workloads, but don't build strict consistency on top of it.
  • Inserting the same data twice won't duplicate rows: Replicated tables deduplicate by the hash of the inserted block, over a window of the last 10,000 blocks (replicated_deduplication_window).

A replica's lag is the length of its queue over that shared log — exactly what system.replication_queue shows. The mechanics fit into a couple of dozen lines:

live example

import java.util.LinkedHashSet;
import java.util.List;
import java.util.Set;

public class KeeperLogDemo {
    record Entry(String type, String part) {}

    public static void main(String[] args) {
        List<Entry> log = List.of(
                new Entry("GET_PART", "202406_1"),
                new Entry("GET_PART", "202406_2"),
                new Entry("GET_PART", "202406_3"),
                new Entry("DROP_TABLE", "-"));

        Set<String> replica = new LinkedHashSet<>();
        int applied = 0;
        for (Entry entry : log) {
            switch (entry.type()) {
                case "GET_PART" -> replica.add(entry.part());
                case "DROP_TABLE" -> replica.clear();
            }
            applied++;
            System.out.println(entry.type() + " " + entry.part()
                    + " -> parts on the replica " + replica
                    + ", lag " + (log.size() - applied));
        }
    }
}
Run

Running examples is part of paid access. There the same code runs inside the article: editor, run and check next to the paragraph. Three free days →

The last entry is the flip side of that design: to a replica, DROP TABLE is the same kind of log entry as fetching a part.

The minimal production topology: 2 replicas + 3 Keeper nodes (Keeper can be colocated with ClickHouse nodes on small clusters). A single node without replication is acceptable if the data can be reloaded from Kafka or PostgreSQL; if ClickHouse is the only place the events are stored, a single node won't do.

Sharding: later than you think

A typical mistake is to shard right away. A single ClickHouse server handles terabytes; as long as vertical growth and a good schema keep up, you don't need sharding. You need it when the data physically no longer fits on one machine, or when a single query saturates the CPU of one server.

The mechanics: data is split across shards, and a router table sits on top:

ENGINE = Distributed(cluster, db, local_table, sharding_key)

A query against a Distributed table fans out across the shards and is assembled on the initiator. Inserts go either through the Distributed table (which spreads them by key) or directly into the shards' local tables (more reliable under load, with less intermediate buffering).

Choosing the sharding_key: even distribution matters more than meaning. cityHash64(order_id) distributes evenly; sharding "by region" gives you a hot shard on a large region.

TTL: when old data is expensive to keep

Analytical data ages, and keeping raw events forever is costly. TTL solves this declaratively:

ALTER TABLE order_events
    MODIFY TTL event_time + INTERVAL 12 MONTH DELETE;

Policy options:

  • DELETE — remove old rows (in the background, during part merges).
  • TO DISK 'cold' / TO VOLUME 'cold' — tiered storage: recent months on fast disks, older ones on cheap disks or S3. This is what other systems call the hot/warm/cold tier.
  • GROUP BY ... SET — reducing precision: for data older than a year, keep daily aggregates instead of individual events.

A common combination: raw events with a 12-month TTL + a materialized view holding aggregates forever. Dashboards run on the aggregates, raw data for pinpoint analysis is available for the last year, and disk stays under control.

Dropping whole partitions is the cheapest way to clean up:

ALTER TABLE order_events DROP PARTITION 202401;

That's another argument for monthly partitions.

Backups

A replica is not a backup: DROP TABLE replicates just as faithfully as INSERT. For real backups there is a built-in mechanism:

BACKUP TABLE analytics.order_events
    TO S3('https://s3.../backups/order_events/2026-06', '...', '...');

RESTORE TABLE analytics.order_events
    FROM S3('https://s3.../backups/order_events/2026-06', '...', '...');

Backups can be incremental: parts are immutable, so only new ones need copying. It is not automatic — BACKUP must name the copy it builds on (SETTINGS base_backup = ...), or every run makes a full one. S3 is the target to aim for.

The second line of defense is reloading from the source: as long as Kafka retains the events or PostgreSQL keeps the history, ClickHouse can be rebuilt. It disappears the moment Kafka's retention is shorter than the depth of data in ClickHouse.

Monitoring through system tables

ClickHouse tells you about itself through SQL queries — all diagnostics live in the system schema:

TableWhat to watchAlert threshold
system.partsNumber of active parts per table-partitionapproaching parts_to_throw_insert (3000 by default) — inserts are about to stall
system.mergesCurrent merges, progressMerges can't keep up with inserts
system.mutationsQueue of ALTER UPDATE/DELETEis_done = 0 for several hours
system.replication_queueReplica lagQueue grows and isn't drained
system.query_logWho read what and how muchQueries reading > N GB or running > N seconds
system.disksFree space< 20% — merges need headroom for a copy of the parts

Exporting metrics to Prometheus is built in — the <prometheus> section in the server config, then the standard dashboards and alerts. A starter set of alerts: too many parts (TOO_MANY_PARTS), replication lag, disk, the share of failed queries, and the duration of background changes.

Common problems and their causes

TOO_MANY_PARTS — insert rejected. The cause is almost always the writing service: small frequent INSERTs, overly granular partitioning, or a cascade of materialized views multiplying the inserts. The fix is batching inserts and revisiting the schema, not bumping the limit.

OOM on a heavy GROUP BY. Aggregation over a high-cardinality column didn't fit into max_memory_usage. The quick remedy is max_bytes_before_external_group_by (spill to disk: slower, but the query survives); the systemic solution is pre-aggregation through materialized views.

Replicas diverged. Look at system.replication_queue and the Keeper logs: most often the cause is the network to Keeper or a full disk. A replica is recovered with these commands:

SYSTEM RESTART REPLICA table_name;
SYSTEM RESTORE REPLICA table_name;

Replication will backfill the data on its own.

Mutations are stuck. The ALTER ... DELETE queue (for example, deleting data on regulator demand) is blocked by a lack of disk or an error in one mutation. A stuck mutation can be cancelled via KILL MUTATION, after which you investigate the cause.

The killer query. SELECT * over two years with no filter. The defense is user profiles: max_execution_time, max_memory_usage, max_rows_to_read for read-only users, and rate quotas. You set these limits before the incident.

Versions and upgrades

ClickHouse ships monthly; for production systems, use the LTS releases (two per year, one year of support). Upgrading a cluster is done replica by replica, without downtime: a replica is taken out, upgraded, catches up on the replication queue, then the next one. Before a major upgrade, run your production queries on a test environment: the optimizer and setting behavior change more noticeably between LTS versions than in PostgreSQL.

In short

  • Replication is at the table level, engine ReplicatedMergeTree; Keeper coordinates it, and the minimal topology is 2 replicas + 3 Keeper nodes.
  • A replica's lag is the length of its queue over the Keeper log; replicas download parts from each other directly.
  • Sharding is needed later than you think — a single server handles terabytes; add it when vertical growth is exhausted.
  • TTL manages the life of the data: deletion, moving to a cold disk, or reducing precision.
  • A replica is not a backup; DROP TABLE replicates. The backup is BACKUP TO S3.
  • All diagnostics are SQL through system.* tables and metric export to Prometheus is built in; the most common outage, TOO_MANY_PARTS, is solved by batching inserts, not by bumping the limit.