Creating a ClickHouse cluster - Part II: Replication
Tags: clickhouse, nosql, docker, cluster
Published: 2021-07-07 15:16:42

In the [previous article](https://dev.to/zergon321/creating-a-clickhouse-cluster-part-i-sharding-4j20) I showed how to run **ClickHouse** in cluster mode using only sharding. It's enough for load distribution, but we also need to ensure fault tolerance via replication.

## ZooKeeper

**Apache ZooKeeper** is a coordination service which provides distributed synchronization of config information across nodes. It's obligatory to have a cluster or a single node of it (above 3.4.5) if you want to enable replication on your **ClickHouse** cluster.

It's pretty easy to launch several nodes of **ZooKeeper** and assemble them into a cluster. Here's the configuration snippet for `docker-compose.yml`:

```yaml
  zk-1:
    container_name: zk_1
    image: bitnami/zookeeper:3.6.1
    environment:
      - ZOO_SERVER_ID=1
      - ALLOW_ANONYMOUS_LOGIN=yes
      - ZOO_SERVERS=zk-1:2888:3888,zk-2:2888:3888

  zk-2:
    container_name: zk_2
    image: bitnami/zookeeper:3.6.1
    environment:
      - ZOO_SERVER_ID=2
      - ALLOW_ANONYMOUS_LOGIN=yes
      - ZOO_SERVERS=zk-1:2888:3888,zk-2:2888:3888
```

And here's my configuration for a single node:

```yaml
  ch-zookeeper:
    container_name: ch_zookeeper
    image: zookeeper:3.6.1
    ports:
      - 2181:2181
```

Pay attention that **ClickHouse** can work with **ZooKeeper** cluster only if its nodes are based on the [Docker image by bitnami](https://hub.docker.com/r/bitnami/zookeeper).

**ClickHouse** does about 10 write operations on **ZooKeeper** for each `INSERT` query.

Now you need to tell all the **ClickHouse** nodes about your **ZooKeeper** cluster. For this, the config file has a `zookeeper` section:

```xml
    <zookeeper>
        <node index="1">
            <host>zk-1</host>
            <port>2181</port>
        </node>
        <node index="2">
            <host>zk-2</host>
            <port>2181</port>
        </node>
    </zookeeper>
```

## Cluster configuration

I will use 1 master with 2 shards, 2 replicas for each shard. Here's the deployments configuration:

```yaml
  ch-master:
    container_name: ch_master
    image: yandex/clickhouse-server:19.14.13.4
    depends_on:
      - ch-zookeeper
      - ch-sub-1
      - ch-sub-2
      - ch-sub-3
      - ch-sub-4
    ports:
      - 9000:9000
      - 8123:8123
    volumes:
      - type: volume
        source: ch-master-data
        target: /var/lib/clickhouse
      - type: volume
        source: ch-master-logs
        target: /var/log/clickhouse-server
      - ./master-config.xml:/etc/clickhouse-server/config.xml

  ch-sub-1:
    container_name: ch_sub_1
    image: yandex/clickhouse-server:19.14.13.4
    depends_on:
      - ch-zookeeper
    ports:
      - 9001:9000
      - 8124:8123
    volumes:
      - type: volume
        source: ch-sub-1-data
        target: /var/lib/clickhouse
      - type: volume
        source: ch-sub-1-logs
        target: /var/log/clickhouse-server
      - ./sub-1-config.xml:/etc/clickhouse-server/config.xml

  ch-sub-2:
    container_name: ch_sub_2
    image: yandex/clickhouse-server:19.14.13.4
    depends_on:
      - ch-zookeeper
    ports:
      - 9002:9000
      - 8125:8123
    volumes:
      - type: volume
        source: ch-sub-2-data
        target: /var/lib/clickhouse
      - type: volume
        source: ch-sub-2-logs
        target: /var/log/clickhouse-server
      - ./sub-2-config.xml:/etc/clickhouse-server/config.xml

  ch-sub-3:
    container_name: ch_sub_3
    image: yandex/clickhouse-server:19.14.13.4
    depends_on:
      - ch-zookeeper
    ports:
      - 9003:9000
      - 8126:8123
    volumes:
      - type: volume
        source: ch-sub-3-data
        target: /var/lib/clickhouse
      - type: volume
        source: ch-sub-3-logs
        target: /var/log/clickhouse-server
      - ./sub-3-config.xml:/etc/clickhouse-server/config.xml

  ch-sub-4:
    container_name: ch_sub_4
    image: yandex/clickhouse-server:19.14.13.4
    depends_on:
      - ch-zookeeper
    ports:
      - 9004:9000
      - 8127:8123
    volumes:
      - type: volume
        source: ch-sub-4-data
        target: /var/lib/clickhouse
      - type: volume
        source: ch-sub-4-logs
        target: /var/log/clickhouse-server
      - ./sub-4-config.xml:/etc/clickhouse-server/config.xml

volumes:
  ch-master-data:
  ch-master-logs:
  ch-sub-1-data:
  ch-sub-1-logs:
  ch-sub-2-data:
  ch-sub-2-logs:
  ch-sub-3-data:
  ch-sub-3-logs:
  ch-sub-4-data:
  ch-sub-4-logs:
```

The `remote_servers` section in `master-config.xml` will now look like this:

```xml
    <remote_servers>
        <example_cluster>
            <shard>
                <internal_replication>true</internal_replication>
                <replica>
                    <host>ch-sub-1</host>
                    <port>9000</port>
                </replica>
                <replica>
                    <host>ch-sub-2</host>
                    <port>9000</port>
                </replica>
            </shard>
            <shard>
                <internal_replication>true</internal_replication>
                <replica>
                    <host>ch-sub-3</host>
                    <port>9000</port>
                </replica>
                <replica>
                    <host>ch-sub-4</host>
                    <port>9000</port>
                </replica>
            </shard>
        </example_cluster>
    </remote_servers>
```

As you can see, the `internal_replication` parameter is now specified for each shard. It's set to `true` so the replicas themselves will send all their data to other replicas of the shard instead of doing it through the master node. If the parameter is set to `false`, the master node will replicate all the data on it's own which may lead to an inconsistent state when all the replicas contain different data.

As for configuration files for subordinate nodes, a new section must be defined in them in order to enable replication. This section is called `macros`.

```xml
    <macros>
        <shard>01</shard>
        <replica>ch-sub-1</replica>
    </macros>
```

It contains parameters that will be accessible through **SQL** queries related to the node. In the above source, `shard` and `replica` are parameters defined by the user, and `"01"` and `"ch-sub-1"` are their corresponding values. The first parameter is a cluster identifier and the second one is a replica name which coincides with the domain name specified in `docker-compose.yml`.

After all the config files are set up, we can finally use scripts to create a cluster and run it. When all the database nodes are up and running, we should first execute our **Python** scripts for subordinate nodes. All of them look like this:

```python
from clickhouse_driver import Client
from datetime import datetime

if __name__ == "__main__":
    client = Client("127.0.0.1", port="9001")

    client.execute("CREATE DATABASE IF NOT EXISTS billing")

    client.execute(r'''CREATE TABLE IF NOT EXISTS billing.transactions(
                      timestamp DateTime,
                      currency String,
                      value Float64)
                      ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/billing.transactions', '{replica}')
                      PARTITION BY currency
                      ORDER BY timestamp''')
```

As you can see, the subordinate table now uses `ReplicatedMergeTree` engine. Its constructor takes the path to the table records in **ZooKeeper** as the first parameter and the replica name as the second parameter. The path to the table in **ZooKeeper** should be unique. All the parameters in `{}` are taken from the aforementioned `macros` section of the replica config file.

When all the subordinate tables are created, it's time to create a master table. There's no difference from the previous case when only sharding was utilized:

```python
from clickhouse_driver import Client
from datetime import datetime

if __name__ == "__main__":
    client = Client("127.0.0.1", port="9000")

    client.execute("CREATE DATABASE IF NOT EXISTS billing")

    client.execute('''CREATE TABLE IF NOT EXISTS billing.transactions(
                      timestamp DateTime,
                      currency String,
                      value Float64)
                      ENGINE = Distributed(example_cluster, billing, transactions, rand())''')
```

If you set up all the things properly, you will get a working **ClickHouse** cluster with replication enabled. The shard is alive if at least one of its replicas is up. Table replication strengthens fault tolerance of the cluster.

All the necessary source code is [here](https://github.com/zergon321/clickhouse-replication).