Day 74/100 - Cross-Datacenter Replication in ClickHouse®

Why Use Cross-Datacenter Replication?

Replicating data across multiple data centers offers several operational and business advantages. Some of the primary benefits include:

• High availability
• Disaster recovery
• Regional redundancy
• Business continuity
• Reduced query latency for geographically distributed users
• Improved fault tolerance
• Maintenance with minimal downtime
• Increased read scalability

Cross-datacenter replication is commonly used in industries such as:

• Financial services
• Telecommunications
• E-commerce
• Healthcare
• SaaS platforms
• Media and streaming services

How Replication Works in ClickHouse®

A single-node ClickHouse deployment has one major limitation-if the server becomes unavailable, so does access to your data. Replication addresses this challenge by maintaining synchronized copies of data across multiple nodes.

ClickHouse replication is powered by the ReplicatedMergeTree table engine. Instead of copying entire tables repeatedly, ClickHouse replicates only the newly created data parts. Metadata about these parts is coordinated using ClickHouse Keeper, while each replica maintains its own local copy of the data. This architecture minimizes network traffic while keeping replicas synchronized efficiently.

Some key benefits include:

• Fault tolerance-if one replica fails, others continue serving queries.
• High availability with no single point of failure.
• Read scalability by distributing SELECT queries across replicas.
• Zero-downtime maintenance by taking individual nodes offline without interrupting service.

Cluster Overview

For this example, consider a cluster consisting of one shard with three replicas distributed across three different nodes.

Each node stores a complete copy of the data. If any single node becomes unavailable, the remaining replicas continue processing queries without data loss. The overall architecture looks like this:

Application
    │
    ▼
Distributed Table
    │
    ┌───────────────┼───────────────┐
    ▼               ▼               ▼
  Node 1          Node 2          Node 3
Replicated      Replicated      Replicated
 MergeTree       MergeTree       MergeTree
    │               │               │
    └───────────────┼───────────────┘
                    ▼
          ClickHouse® Keeper

Core Components

Cross-datacenter replication relies on three primary components working together.

1. ReplicatedMergeTree

This is the storage engine responsible for replication. Each replica stores its own copy of the table while automatically synchronizing newly inserted data with other replicas.

2. ClickHouse Keeper

ClickHouse Keeper is the coordination service responsible for managing replication metadata. It tracks:

• Available replicas
• Newly created data parts
• Replication queues
• Leader election
• Replica synchronization

Keeper uses the Raft consensus algorithm to ensure fault-tolerant coordination.

3. Distributed Table

Applications should interact with a Distributed table instead of local replicated tables. The Distributed table automatically routes queries and inserts to the appropriate replicas while hiding the underlying cluster topology.

Configuration Overview

Setting up replication requires several configuration files under:

/etc/clickhouse-server/config.d/

Each file serves a different purpose.

After updating these configuration files, restart ClickHouse on every node before creating replicated tables.

Important Notes:

• cluster.xml is identical across all nodes.
• zookeeper.xml is identical across all nodes.
• keeper.xml contains node-specific server_id values.
• macros.xml contains node-specific replica names.

Creating Replicated Tables

Once the cluster is configured and ClickHouse has been restarted on every node, create the database and tables.

Step 1 - Create the Database

CREATE DATABASE analytics ON CLUSTER cluster_1S_3R;

Step 2 - Create the Local Replicated Table

CREATE TABLE analytics.orders_local ON CLUSTER cluster_1S_3R
(
    order_id UInt32,
    customer String,
    amount Float64,
    order_date Date
)
ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/orders_local', '{replica}')
ORDER BY order_id;

The placeholders {shard} and {replica} are automatically replaced using values from macros.xml.

Step 3 - Create the Distributed Table

CREATE TABLE analytics.orders_distributed ON CLUSTER cluster_1S_3R
AS analytics.orders_local
ENGINE = Distributed(cluster_1S_3R, analytics, orders_local, rand());

Applications should perform all inserts and queries using this Distributed table.

Verifying Replication

Insert sample records.

INSERT INTO analytics.orders_distributed VALUES
(1, 'Alice', 1200.00, '2024-01-01'),
(2, 'Bob', 450.00, '2024-01-02'),
(3, 'Charlie', 890.00, '2024-01-03'),
(4, 'Diana', 670.00, '2024-01-04');

Now query the local table from each node.

SELECT * FROM analytics.orders_local ORDER BY order_id;

Expected output on every replica:

Seeing identical data across all replicas confirms that replication is functioning correctly.

How Replication Works Internally

When an application inserts data into the Distributed table, ClickHouse performs the following sequence automatically:

Application
    │
    ▼
Distributed Table
    │
    ▼
Selected Replica
    │
    ▼
Writes Data Part
    │
    ▼
ClickHouse Keeper Updates Metadata
    │
    ▼
Remaining Replicas Fetch Data
    │
    ▼
Replication Complete

The process consists of these steps:

1. The Distributed table routes the insert to one replica.
2. That replica writes a new data part locally.
3. ClickHouse Keeper records the new data part.
4. Other replicas detect the update.
5. Replicas download the new part.
6. Keeper confirms successful synchronization.

Every replica now contains identical data. This synchronization occurs automatically without requiring manual intervention.

Monitoring Replication

Monitoring replica health is essential for production clusters. Query the system.replicas table.

SELECT
    database,
    table,
    replica_name,
    is_leader,
    is_readonly,
    total_replicas,
    active_replicas,
    queue_size
FROM system.replicas
WHERE table = 'orders_local';

Example output:

Healthy replication typically shows:

• active_replicas = total_replicas
• is_readonly = 0
• queue_size = 0

Checking Replication Errors

To identify replication failures:

SELECT
    database,
    table,
    replica_name,
    last_exception
FROM system.replicas
WHERE last_exception != '';

Any returned rows indicate replication issues that require investigation.

Verifying ClickHouse Keeper

You can verify Keeper availability using:

echo "ruok" | nc 10.x.x.1 9181

Expected response: imok

If Keeper does not respond, replication will eventually stop because replicas can no longer coordinate metadata.

Common Issues and Solutions

Performance and Operational Best Practices

To build a reliable replicated ClickHouse deployment:

• Deploy an odd number of Keeper nodes (3 or 5) to maintain quorum.
• Assign every Keeper node a unique server_id.
• Give every replica a unique name in macros.xml.
• Always execute DDL statements using ON CLUSTER.
• Perform inserts through the Distributed table instead of local tables.
• Monitor system.replicas regularly for replication lag.
• Restart cluster nodes one at a time.
• Ensure stable, low-latency connectivity between data centers.
• Batch inserts to reduce replication overhead.
• Continuously monitor Keeper health and replication queues.

When Should You Use Cross-Datacenter Replication?

Cross-datacenter replication is an excellent choice when your environment requires:

• High availability
• Disaster recovery
• Multi-region deployments
• Read scalability
• Business continuity
• Low-latency regional analytics
• Fault-tolerant analytical platforms

Organizations operating globally often combine replicated tables with distributed queries to provide seamless analytics regardless of where users connect.

Conclusion

Cross-datacenter replication enables ClickHouse® to maintain synchronized copies of data across geographically distributed environments, improving availability, fault tolerance, and disaster recovery capabilities. By combining ReplicatedMergeTree, ClickHouse Keeper, and Distributed tables, organizations can build resilient analytical platforms that continue serving queries even when individual nodes or data centers become unavailable.

Although network latency and bandwidth should be considered when replicating across regions, following best practices such as batching inserts, maintaining a healthy Keeper cluster, monitoring replication queues, and using Distributed tables for application traffic helps ensure reliable and efficient synchronization. For organizations requiring both high-performance analytics and resilient multi-region deployments, ClickHouse provides a scalable and robust replication architecture that minimizes operational complexity while maximizing availability.