Replication + Scaling

Notice that it is identical to the query used in the original `CREATE` statement of the
[UK property prices](/docs/getting-started/example-datasets/uk-price-paid) example dataset tutorial,
except for the `ON CLUSTER` clause and use of the `ReplicatedMergeTree` engine.

The `ON CLUSTER` clause is designed for distributed execution of DDL (Data Definition Language)
queries such as `CREATE`, `DROP`, `ALTER`, and `RENAME`, ensuring that these
schema changes are applied across all nodes in a cluster.

The [`ReplicatedMergeTree`](https://clickhouse.com/docs/engines/table-engines/mergetree-family/replication#converting-from-mergetree-to-replicatedmergetree)
engine works just as the ordinary `MergeTree` table engine, but it will also replicate the data.
It requires two parameters to be specified:

- `zoo_path`: The Keeper/ZooKeeper path to the table's metadata.
- `replica_name`: The table's replica name.

  

The `zoo_path` parameter can be set to anything you choose, although it is recommended to follow
the convention of using prefix

```
/clickhouse/tables/{shard}/{database}/{table}
```

where:

- `{database}` and `{table}` will be replaced automatically.
- `{shard}` and `{replica}` are macros which were [defined](#macros-config-explanation)
  previously in the `config.xml` file of each ClickHouse node.

You can run the query below from each host's client to confirm that the table has been created across the cluster:

```
SHOW TABLES IN uk;
```

```
   ┌─name────────────────┐
1. │ uk_price_paid_local │
   └─────────────────────┘
```

Insert data into a distributed table

## Insert data into a distributed table[​](#inserting-data-using-distributed "Direct link to Insert data into a distributed table")

To insert data into the table, `ON CLUSTER` can't be used as it does
not apply to DML (Data Manipulation Language) queries such as `INSERT`, `UPDATE`,
and `DELETE`. To insert data, it is necessary to make use of the
[`Distributed`](/docs/engines/table-engines/special/distributed) table engine.
As you learned in the [guide](/docs/architecture/horizontal-scaling) for setting up a cluster with 2 shards and 1 replica, distributed tables are tables which have access to shards located on different
hosts and are defined using the `Distributed` table engine.
The distributed table acts as the interface across all the shards in the cluster.

From any of the host clients, run the following query to create a distributed table
using the existing replicated table we created in the previous step:

```
CREATE TABLE IF NOT EXISTS uk.uk_price_paid_distributed
ON CLUSTER cluster_2S_2R
ENGINE = Distributed('cluster_2S_2R', 'uk', 'uk_price_paid_local', rand());
```

On each host you will now see the following tables in the `uk` database:

```
   ┌─name──────────────────────┐
1. │ uk_price_paid_distributed │
2. │ uk_price_paid_local       │
   └───────────────────────────┘
```

Data can be inserted into the `uk_price_paid_distributed` table from any of the
host clients using the following query:

```
INSERT INTO uk.uk_price_paid_distributed
SELECT
    toUInt32(price_string) AS price,
    parseDateTimeBestEffortUS(time) AS date,
    splitByChar(' ', postcode)[1] AS postcode1,
    splitByChar(' ', postcode)[2] AS postcode2,
    transform(a, ['T', 'S', 'D', 'F', 'O'], ['terraced', 'semi-detached', 'detached', 'flat', 'other']) AS type,
    b = 'Y' AS is_new,
    transform(c, ['F', 'L', 'U'], ['freehold', 'leasehold', 'unknown']) AS duration,
    addr1,
    addr2,
    street,
    locality,
    town,
    district,
    county
FROM url(
    'http://prod1.publicdata.landregistry.gov.uk.s3-website-eu-west-1.amazonaws.com/pp-complete.csv',
    'CSV',
    'uuid_string String,
    price_string String,
    time String,
    postcode String,
    a String,
    b String,
    c String,
    addr1 String,
    addr2 String,
    street String,
    locality String,
    town String,
    district String,
    county String,
    d String,
    e String'
) SETTINGS max_http_get_redirects=10;
```

Run the following query to confirm that the data inserted has been evenly distributed
across the nodes of our cluster:

```
SELECT count(*)
FROM uk.uk_price_paid_distributed;

SELECT count(*) FROM uk.uk_price_paid_local;
```

```
   ┌──count()─┐
1. │ 30212555 │ -- 30.21 million
   └──────────┘

   ┌──count()─┐
1. │ 15105983 │ -- 15.11 million
   └──────────┘
```

## Conclusion[​](#conclusion "Direct link to Conclusion")

The advantage of this cluster topology with 2 shards and 2 replicas is that it provides both scalability and fault tolerance.
Data is distributed across separate hosts, reducing storage and I/O requirements per node, while queries are processed in parallel across both shards for improved performance and memory efficiency.
Critically, the cluster can tolerate the loss of one node and continue serving queries without interruption, as each shard has a backup replica available on another node.

The main disadvantage of this cluster topology is the increased storage overhead—it requires twice the storage capacity compared to a setup without replicas, as each shard is duplicated.
Additionally, while the cluster can survive a single node failure, losing two nodes simultaneously may render the cluster inoperable, depending on which nodes fail and how shards are distributed.
This topology strikes a balance between availability and cost, making it suitable for production environments where some level of fault tolerance is required without the expense of higher replication factors.

To learn how ClickHouse Cloud processes queries, offering both scalability and fault-tolerance, see the section ["Parallel Replicas"](/docs/deployment-guides/parallel-replicas).

- [Prerequisites](#prerequisites)- [Set up directory structure and test environment](#set-up)- [Configure ClickHouse nodes](#configure-clickhouse-servers)
      - [Server setup](#server-setup)- [User configuration](#cluster-configuration)- [Configure ClickHouse Keeper](#configure-clickhouse-keeper-nodes)
        - [Keeper setup](#configuration-explanation)- [Test the setup](#test-the-setup)- [Create a database](#creating-a-database)- [Create a table on the cluster](#creating-a-table)- [Insert data into a distributed table](#inserting-data-using-distributed)- [Conclusion](#conclusion)

Was this page helpful?