Scaling

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.

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 │
   └─────────────────────┘
```

Before we insert the UK price paid data, let's perform a quick experiment to see
what happens when we insert data into an ordinary table from either host.

Create a test database and table with the following query from either host:

```
CREATE DATABASE IF NOT EXISTS test ON CLUSTER cluster_2S_1R;
CREATE TABLE test.test_table ON CLUSTER cluster_2S_1R
(
    `id` UInt64,
    `name` String
)
ENGINE = MergeTree()
ORDER BY id;
```

Now from `clickhouse-01` run the following `INSERT` query:

```
INSERT INTO test.test_table (id, name) VALUES (1, 'Clicky McClickface');
```

Switch over to `clickhouse-02` and run the following `INSERT` query:

```
INSERT INTO test.test_table (id, name) VALUES (1, 'Alexey Milovidov');
```

Now from `clickhouse-01` or `clickhouse-02` run the following query:

```
-- from clickhouse-01
SELECT * FROM test.test_table;
--   ┌─id─┬─name───────────────┐
-- 1.│  1 │ Clicky McClickface │
--   └────┴────────────────────┘

--from clickhouse-02
SELECT * FROM test.test_table;
--   ┌─id─┬─name───────────────┐
-- 1.│  1 │ Alexey Milovidov   │
--   └────┴────────────────────┘
```

You will notice that unlike with a `ReplicatedMergeTree` table only the row that was inserted into the table on that
particular host is returned and not both rows.

To read the data across the two shards, we need an interface which can handle queries
across all the shards, combining the data from both shards when we run select queries
on it or inserting data to both shards when we run insert queries.

In ClickHouse this interface is called a **distributed table**, which we create using
the [`Distributed`](/docs/engines/table-engines/special/distributed) table engine. Let's take a look at how it works.

Create a distributed table

## Create a distributed table[​](#create-distributed-table "Direct link to Create a distributed table")

Create a distributed table with the following query:

```
CREATE TABLE test.test_table_dist ON CLUSTER cluster_2S_1R AS test.test_table
ENGINE = Distributed('cluster_2S_1R', 'test', 'test_table', rand())
```

In this example, the `rand()` function is chosen as the sharding key so that
inserts are randomly distributed across the shards.

Now query the distributed table from either host, and you will get back
both of the rows which were inserted on the two hosts, unlike in our previous example:

```
SELECT * FROM test.test_table_dist;
```

```
   ┌─id─┬─name───────────────┐
1. │  1 │ Alexey Milovidov   │
2. │  1 │ Clicky McClickface │
   └────┴────────────────────┘
```

Let's do the same for our UK property prices data. From any of the host clients,
run the following query to create a distributed table using the existing table
we created previously with `ON CLUSTER`:

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

Insert data into a distributed table

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

Now connect to either of the hosts and insert the data:

```
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;
```

Once the data is inserted, you can check the number of rows using the distributed
table:

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

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

If you run the following query on either host you will see that the data has been
more or less evenly distributed across the shards (keeping in mind the choice of which
shard to insert into was set with `rand()` so results may differ for you):

```
-- from clickhouse-01
SELECT count(*)
FROM uk.uk_price_paid_local
--    ┌──count()─┐
-- 1. │ 15107353 │ -- 15.11 million
--    └──────────┘

--from clickhouse-02
SELECT count(*)
FROM uk.uk_price_paid_local
--    ┌──count()─┐
-- 1. │ 15105202 │ -- 15.11 million
--    └──────────┘
```

What will happen if one of the hosts fails? Let's simulate this by shutting down
`clickhouse-01`:

```
docker stop clickhouse-01
```

Check that the host is down by running:

```
docker-compose ps
```

```
NAME                   IMAGE                                        COMMAND            SERVICE                CREATED          STATUS          PORTS
clickhouse-02          clickhouse/clickhouse-server:latest          "/entrypoint.sh"   clickhouse-02          X minutes ago    Up X minutes    127.0.0.1:8124->8123/tcp, 127.0.0.1:9001->9000/tcp
clickhouse-keeper-01   clickhouse/clickhouse-keeper:latest-alpine   "/entrypoint.sh"   clickhouse-keeper-01   X minutes ago    Up X minutes    127.0.0.1:9181->9181/tcp
clickhouse-keeper-02   clickhouse/clickhouse-keeper:latest-alpine   "/entrypoint.sh"   clickhouse-keeper-02   X minutes ago    Up X minutes    127.0.0.1:9182->9181/tcp
clickhouse-keeper-03   clickhouse/clickhouse-keeper:latest-alpine   "/entrypoint.sh"   clickhouse-keeper-03   X minutes ago    Up X minutes    127.0.0.1:9183->9181/tcp
```

Now from `clickhouse-02` run the same select query we ran before on the distributed
table:

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

```
Received exception from server (version 25.5.2):
Code: 279. DB::Exception: Received from localhost:9000. DB::Exception: All connection tries failed. Log:

Code: 32. DB::Exception: Attempt to read after eof. (ATTEMPT_TO_READ_AFTER_EOF) (version 25.5.2.47 (official build))
Code: 209. DB::NetException: Timeout: connect timed out: 192.168.7.1:9000 (clickhouse-01:9000, 192.168.7.1, local address: 192.168.7.2:37484, connection timeout 1000 ms). (SOCKET_TIMEOUT) (version 25.5.2.47 (official build))
#highlight-next-line
Code: 198. DB::NetException: Not found address of host: clickhouse-01: (clickhouse-01:9000, 192.168.7.1, local address: 192.168.7.2:37484). (DNS_ERROR) (version 25.5.2.47 (official build))

: While executing Remote. (ALL_CONNECTION_TRIES_FAILED)
```

Unfortunately, our cluster isn't fault-tolerant. If one of the hosts fails, the
cluster is considered unhealthy and the query fails compared to the replicated
table we saw in the [previous example](/docs/architecture/replication) for which
we were able to insert data even when one of the hosts failed.

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

The advantage of this cluster topology is that data gets distributed across
separate hosts and uses half the storage per node. More importantly, queries
are processed across both shards, which is more efficient in terms of memory
utilization and reduces I/O per host.

The main disadvantage of this cluster topology is, of course, that losing one of
the hosts renders us unable to serve queries.