Setup ClickHouse Cluster Replication with Zookeeper

ClickHouse Replication

Introduction

ClickHouse is a powerful and versatile open-source columnar database management system known for its fast performance and high scalability. If you’re looking to build your own ClickHouse cluster, there are several options available, such as using AWS EKS Service, Altinity’s Kubernetes Operator, or Dockerization. However, in this comprehensive blog post, we will walk you through the steps to host your own ClickHouse cluster on Amazon EC2, which can also be applied to other cloud providers like GCP or Azure, or even your own data center. By following these steps, you’ll be able to set up a robust and efficient ClickHouse cluster to meet your specific requirements.

 

Environment

The current setup has hosted the environment on the AWS cloud, utilizing a public subnet for testing. It’s advisable to host the database on a private subnet with restricted access in a production environment. This advanced practice adds an extra layer of security and effectively mitigates potential security risks from unauthorized access.

Operating System – Ubuntu 22.04.2 LTS

Role of ZooKeeper in ClickHouse Cluster – Apache ZooKeeper is a distributed coordination service often used in distributed systems. The role of ZooKeeper in ClickHouse replication involves managing the coordination and synchronization of distributed ClickHouse instances for replication purposes and Failover Handling. It stores and manages configuration information for ClickHouse replicas, such as replica metadata, replica status, and replica configuration settings. ClickHouse replicas can use this information to determine the state of the replication setup and make decisions accordingly.

Two-node replicated cluster – With replication, we can achieve exceptional benefits such as high availability, efficient load balancing, and seamless handling of upgrades and migrations in ClickHouse. Replication operates at the table level, giving you the flexibility to selectively replicate your data across multiple nodes. Asynchronous replication ensures that data consistency is eventually achieved across the entire cluster, making it a bidirectional multi-master replication solution.

Broad Level Action

We will perform the following broad-level actions:

  1. Set hostnames
  2. Install ClickHouse on two nodes.
  3. Install Zookeeper on a single node.
  4. Configure changes on the Zookeeper side.
  5. Perform configuration on ClickHouse nodes to create a cluster.
  6. Test the connection.
  7. Set up a sample database and table for replication.

Set hostnames for all servers

By setting the hostnames and allowing remote connection by name in the /etc/hosts file, and using the hostnamectl command, you ensure that the nodes can be identified by name and communicate with each other in the ClickHouse cluster.

Make the below changes and reboot the servers for the changes to take effect.

root@ip-10-0-11-49:~# cat /etc/hosts
10.0.11.49 zookeeper1

root@ip-10-0-14-42:~# cat /etc/hosts
10.0.11.49 zookeeper1
10.0.14.42 clickhouse1
10.0.12.141 clickhouse2

root@ip-10-0-12-141:~# cat /etc/hosts
10.0.11.49 zookeeper1
10.0.14.42 clickhouse1
10.0.12.141 clickhouse2

root@ip-10-0-14-42:~# hostnamectl set-hostname clickhouse1
root@ip-10-0-12-141:~# hostnamectl set-hostname clickhouse2

root@ip-10-0-11-49:~# reboot
root@ip-10-0-14-42:~# reboot
root@ip-10-0-12-141:~# reboot

Install and Configure Zookeeper

Requirement: Same or greater than version 3.4.5 for ZooKeeper

We’ve installed the most current version from the Ubuntu repository, Zookeeper 3.4.13, using the following command.

This is a test example we have used one ZooKeeper node however, to create a resilient and fault-tolerant setup for Zookeeper, it is advisable to utilize a three-node cluster in accordance with the quorum principle. You can follow these steps to establish a cohesive three-node cluster for enhanced reliability.

root@ip-10-0-11-49:~#  sudo apt-get update

root@ip-10-0-11-49:~#  sudo apt-get install zookeeper netcat

root@ip-10-0-11-49:~# sudo cat /etc/zookeeper/conf/myid
1

root@ip-10-0-11-49:~# sudo cat /etc/zookeeper/conf/zoo.cfg 
tickTime=2000
initLimit=20
syncLimit=10
dataDir=/var/lib/zookeeper
clientPort=2181
maxSessionTimeout=60000000
maxClientCnxns=2000
server.1=zookeeper1:2888:3888
autopurge.purgeInterval=1
autopurge.snapRetainCount=10
4lw.commands.whitelist=*
preAllocSize=131072
snapCount=3000000

root@ip-10-0-11-49:~# sudo -u zookeeper /usr/share/zookeeper/bin/zkServer.sh start 
ZooKeeper JMX enabled by default 
Using config: /etc/zookeeper/conf/zoo.cfg 
Starting zookeeper ... STARTED

Verify Zookeeper Installation and Connections

We can verify Zookeeper is running on the local host with the following three commands.  The output will be similar to what you see in the examples below.

root@ip-10-0-11-49:~# echo ruok | nc localhost 2181
imok

root@ip-10-0-11-49:~# echo mntr | nc localhost 2181
zk_version 3.4.13-6ubuntu4--1, built on Wed, 23 Mar 2022 13:40:46 +0000
zk_avg_latency 0
zk_max_latency 30
zk_min_latency 0
zk_packets_received 1903379
zk_packets_sent 1903422
zk_num_alive_connections 4
zk_outstanding_requests 0
zk_server_state standalone
zk_znode_count 163
zk_watch_count 10
zk_ephemerals_count 4
zk_approximate_data_size 18826
zk_open_file_descriptor_count 33
zk_max_file_descriptor_count 1048576
zk_fsync_threshold_exceed_count 0
root@ip-10-0-11-49:~#
root@ip-10-0-11-49:~# echo stat | nc localhost 2181
Zookeeper version: 3.4.13-6ubuntu4--1, built on Wed, 23 Mar 2022 13:40:46 +0000
Clients:
/10.0.12.141:37442[1](queued=0,recved=951619,sent=951641)
/104.152.52.103:55363[1](queued=0,recved=0,sent=0)
/0:0:0:0:0:0:0:1:56028[0](queued=0,recved=1,sent=0)
/10.0.14.42:49638[1](queued=0,recved=951743,sent=951765)

Latency min/avg/max: 0/0/30
Received: 1903411
Sent: 1903454
Connections: 4
Outstanding: 0
Zxid: 0x181
Mode: standalone
Node count: 163
root@ip-10-0-11-49:~#

root@ip-10-0-11-49:~# echo stat | nc 10.0.11.49 2181
Zookeeper version: 3.4.13-6ubuntu4--1, built on Wed, 23 Mar 2022 13:40:46 +0000
Clients:
/10.0.12.141:37442[1](queued=0,recved=951451,sent=951473)
/104.152.52.103:55363[1](queued=0,recved=0,sent=0)
/10.0.14.42:49638[1](queued=0,recved=951572,sent=951594)
/10.0.14.42:54748[0](queued=0,recved=1,sent=0)

Latency min/avg/max: 0/0/30
Received: 1903068
Sent: 1903111
Connections: 4
Outstanding: 0
Zxid: 0x181
Mode: standalone
Node count: 163

root@ip-10-0-11-49:~# echo stat | nc 10.0.11.49 2181
Zookeeper version: 3.4.13-6ubuntu4--1, built on Wed, 23 Mar 2022 13:40:46 +0000
Clients:
/10.0.12.141:37442[1](queued=0,recved=951465,sent=951487)
/104.152.52.103:55363[1](queued=0,recved=0,sent=0)
/10.0.12.141:41738[0](queued=0,recved=1,sent=0)
/10.0.14.42:49638[1](queued=0,recved=951572,sent=951594)

Latency min/avg/max: 0/0/30
Received: 1903083
Sent: 1903126
Connections: 4
Outstanding: 0
Zxid: 0x181
Mode: standalone
Node count: 163
root@clickhouse2:~#

Install and Configure Clickhouse nodes.

Execute ClickHouse installation by adding the build repository to the local system and running the following commands on both DB Nodes.

Note: At the end of the installation script, you may be prompted to set a password for the default user. Do not enter a password.

root@clickhouse1:~# sudo apt update && sudo apt upgrade -y
root@clickhouse1:~# apt-get install -y apt-transport-https ca-certificates dirmngr
root@clickhouse1:~# mktemp -d
root@clickhouse1:~# GNUPGHOME=$(mktemp -d)
root@clickhouse1:~# sudo GNUPGHOME="$GNUPGHOME" gpg --no-default-keyring --keyring /usr/share/keyrings/clickhouse-keyring.gpg --keyserver hkp://keyserver.ubuntu.com:80 --recv-keys 8919F6BD2B48D754
root@clickhouse1:~# sudo rm -r "$GNUPGHOME"
root@clickhouse1:~# sudo chmod +r /usr/share/keyrings/clickhouse-keyring.gpg
root@clickhouse1:~# sudo rm -r "$GNUPGHOME"
root@clickhouse1:~# echo "deb [signed-by=/usr/share/keyrings/clickhouse-keyring.gpg] https://packages.clickhouse.com/deb stable main" | sudo tee \
    /etc/apt/sources.list.d/clickhouse.list
deb [signed-by=/usr/share/keyrings/clickhouse-keyring.gpg] https://packages.clickhouse.com/deb stable main
root@clickhouse1:~# sudo apt-get update
root@clickhouse1:~# sudo apt-get install -y clickhouse-server clickhouse-client

[DO NOT ENTER DEFAULT USER PASSWORD ]

root@clickhouse1:~# service clickhouse-server start

root@clickhouse1:~# clickhouse-client
ClickHouse client version 23.3.1.2823 (official build).
Connecting to localhost:9000 as user default.
Connected to ClickHouse server version 23.3.1 revision 54462.

Warnings:
 * Linux threads max count is too low. Check /proc/sys/kernel/threads-max
 * Available memory at server startup is too low (2GiB).
 * Maximum number of threads is lower than 30000. There could be problems with handling a lot of simultaneous queries.

Execute the same steps on another DB node, i.e. on clickhouse2 and If you have entered the default password already, follow the below steps to clean this out.

root@clickhouse1:~# rm /etc/clickhouse-server/users.d/default-password.xml
root@clickhouse1:~# clickhouse-server start
Server started

Configure Zookeeper for clickhouse1 and clickhouse2

Create a new configuration file /etc/clickhouse-server/config.d/zookeeper.xml for both DB servers,

We only have one Zookeeper node, so we’ll add that to our configuration file. The following settings are the most important ones.

  • <zookeeper><node>: There is one <node> entry for each Zookeeper node. We only have one, so that’s all we need here.
  • <distributed_ddl><path>: This needs to be unique for each cluster. We’re only creating one cluster with our 2  DB nodes.
root@clickhouse1:/etc/clickhouse-server/config.d# cat zookeeper.xml
<yandex>
    <zookeeper>
        <node>
            <host>zookeeper1</host>
            <port>2181</port>
        </node>
        <session_timeout_ms>30000</session_timeout_ms>
        <operation_timeout_ms>10000</operation_timeout_ms>
        <!-- Optional. Chroot suffix. Should exist. -->
        <!-- <root>/path/to/zookeeper/node</root> -->
        <!-- Optional. ZooKeeper digest ACL string. -->
        <!-- <identity>user:password</identity> -->
    </zookeeper>
    <!-- Allow to execute distributed DDL queries (CREATE, DROP, ALTER, RENAME) on cluster. -->
    <!-- Works only if ZooKeeper is enabled. Comment it out if such functionality isn't required. -->
    <distributed_ddl>
        <!-- Path in ZooKeeper to queue with DDL queries -->
        <path>/clickhouse/task_queue/ddl</path>

        <!-- Settings from this profile will be used to execute DDL queries -->
        <!-- <profile>default</profile> -->
    </distributed_ddl>
</yandex>

root@clickhouse2:/etc/clickhouse-server/config.d# cat zookeeper.xml
<yandex>
    <zookeeper>
        <node>
            <host>zookeeper1</host>
            <port>2181</port>
        </node>
        <session_timeout_ms>30000</session_timeout_ms>
        <operation_timeout_ms>10000</operation_timeout_ms>
        <!-- Optional. Chroot suffix. Should exist. -->
        <!-- <root>/path/to/zookeeper/node</root> -->
        <!-- Optional. ZooKeeper digest ACL string. -->
        <!-- <identity>user:password</identity> -->
    </zookeeper>
    <!-- Allow to execute distributed DDL queries (CREATE, DROP, ALTER, RENAME) on cluster. -->
    <!-- Works only if ZooKeeper is enabled. Comment it out if such functionality isn't required. -->
    <distributed_ddl>
        <!-- Path in ZooKeeper to queue with DDL queries -->
        <path>/clickhouse/task_queue/ddl</path>

        <!-- Settings from this profile will be used to execute DDL queries -->
        <!-- <profile>default</profile> -->
    </distributed_ddl>
</yandex>

Macro Settings for clickhouse1 and clickhouse2 Servers

Define the macro configuration by creating the file for macros in the file /etc/clickhouse-server/config.d/macros.xml. This sets how the shards and tables will set the paths of where they store the replicated tables. We have three values we care about

  • Cluster: the name of our cluster will be cluster_demo_ash.
  • Shard: We have just one shard, so we’ll make that number 1.
  • Replica: This value is unique per node and will be the node’s hostname.
root@clickhouse1:/etc/clickhouse-server/config.d# cat macros.xml
<yandex>
    <macros>
       <cluster>cluster_demo_ash</cluster>
        <shard>1</shard>
        <replica>clickhouse1</replica>
    </macros>
</yandex>

root@clickhouse2:/etc/clickhouse-server/config.d# cat macros.xml
<yandex>
<macros>
<cluster>cluster_demo_ash</cluster>
<shard>1</shard>
<replica>clickhouse2</replica>
</macros>
</yandex>

Define Cluster for clickhouse1 and clickhouse2 Servers.

Create new file cluster.xml and define cluster with nodes in shard and replica on all the nodes which are part of the cluster.

root@clickhouse1:~# cat /etc/clickhouse-server/config.d/clusters.xml
<yandex>
    <remote_servers>
        <cluster_demo_ash>
            <shard>
                <replica>
                    <host>clickhouse1</host>
                    <port>9000</port>
                </replica>
                <replica>
                    <host>clickhouse2</host>
                    <port>9000</port>
                </replica>
            </shard>
        </cluster_demo_ash>
    </remote_servers>
</yandex>
root@clickhouse2:~# cat /etc/clickhouse-server/config.d/clusters.xml
<yandex>
    <remote_servers>
        <cluster_demo_ash>
            <shard>
                <replica>
                    <host>clickhouse1</host>
                    <port>9000</port>
                </replica>
                <replica>
                    <host>clickhouse2</host>
                    <port>9000</port>
                </replica>
            </shard>
        </cluster_demo_ash>
    </remote_servers>
</yandex>

Open the remote connection

To enable internode communication within the cluster, make changes to the listen_host configuration to allow network access. Create a new file named listen_host.xml and make the following changes. After that, attempt to establish a remote connection from one ClickHouse node to another.

Note – Restarting the ClickHouse service is required for the listen_host changes to take effect.

root@clickhouse1:~#  cat /etc/clickhouse-server/config.d/listen_host.xml
<clickhouse>
    <listen_host>::</listen_host>
</clickhouse>

root@clickhouse1:~# sudo clickhouse-server start
Server started
root@clickhouse1:~# clickhouse-client --host=clickhouse2
ClickHouse client version 23.3.1.2823 (official build).
Connecting to clickhouse2:9000 as user default.
Connected to ClickHouse server version 23.3.1 revision 54462.

Warnings:
 * Linux threads max count is too low. Check /proc/sys/kernel/threads-max
 * Available memory at server startup is too low (2GiB).
 * Maximum number of threads is lower than 30000. There could be problems with handling a lot of simultaneous queries.

root@clickhouse2:~# cat /etc/clickhouse-server/config.d/listen_host.xml
<clickhouse>
    <listen_host>::</listen_host>
</clickhouse>

root@clickhouse2:~# sudo clickhouse-server start
Server started
root@clickhouse2:~# clickhouse-client --host=clickhouse1
ClickHouse client version 23.3.1.2823 (official build).
Connecting to clickhouse1:9000 as user default.
Connected to ClickHouse server version 23.3.1 revision 54462.

Warnings:
 * Linux threads max count is too low. Check /proc/sys/kernel/threads-max
 * Available memory at server startup is too low (2GiB).
 * Maximum number of threads is lower than 30000. There could be problems with handling a lot of simultaneous queries.

Verify Clickhouse Cluster

Check the status of the cluster by using the system tables system.cluster, which should display entries for all nodes that are part of the specified cluster.

root@clickhouse1:~# clickhouse-client -q "SELECT * FROM system.clusters WHERE cluster='cluster_demo_ash' FORMAT Vertical;"
Row 1:
──────
cluster:                 cluster_demo_ash
shard_num:               1
shard_weight:            1
replica_num:             1
host_name:               clickhouse1
host_address:            10.0.14.42
port:                    9000
is_local:                1
user:                    default
default_database:
errors_count:            0
slowdowns_count:         0
estimated_recovery_time: 0

Row 2:
──────
cluster:                 cluster_demo_ash
shard_num:               1
shard_weight:            1
replica_num:             2
host_name:               clickhouse2
host_address:            10.0.12.141
port:                    9000
is_local:                0
user:                    default
default_database:
errors_count:            0
slowdowns_count:         0
estimated_recovery_time: 0
root@clickhouse1:~#
root@clickhouse1:~#
root@clickhouse1:~# clickhouse-client -q "select * from system.zookeeper where path='/clickhouse/task_queue/'"
ddl		/clickhouse/task_queue/
root@clickhouse1:~#

Create a sample Database and Replicated table for Cluster

With our Zookeeper and ClickHouse servers talking to each other, let’s make a distributed table and test it out.

clickhouse2 :) CREATE DATABASE IF NOT EXISTS Example_DB ON CLUSTER '{cluster}';

CREATE DATABASE IF NOT EXISTS Example_DB ON CLUSTER `{cluster}`

Query id: 50ee6543-4937-4186-8768-16b073277d18

┌─host────────┬─port─┬─status─┬─error─┬─num_hosts_remaining─┬─num_hosts_active─┐
│ clickhouse2 │ 9000 │      0 │       │                   1 │                0 │
│ clickhouse1 │ 9000 │      0 │       │                   0 │                0 │
└─────────────┴──────┴────────┴───────┴─────────────────────┴──────────────────┘

2 rows in set. Elapsed: 0.113 sec.

clickhouse2 :) CREATE TABLE Example_DB.product ON CLUSTER '{cluster}'
               (
               created_at DateTime,
               product_id UInt32,
               category UInt32
               ) ENGINE = ReplicatedMergeTree('/clickhouse/tables/{cluster}/{shard}/Example_DB/product', '{replica}')
               PARTITION BY toYYYYMM(created_at)
               ORDER BY (product_id, toDate(created_at), category)
               SAMPLE BY category;


CREATE TABLE Example_DB.product ON CLUSTER `{cluster}`
(
    `created_at` DateTime,
    `product_id` UInt32,
    `category` UInt32
)
ENGINE = ReplicatedMergeTree('/clickhouse/tables/{cluster}/{shard}/Example_DB/product', '{replica}')
PARTITION BY toYYYYMM(created_at)
ORDER BY (product_id, toDate(created_at), category)
SAMPLE BY category

Query id: b041cfa2-455d-4959-b88d-1554467c9de3

┌─host────────┬─port─┬─status─┬─error─┬─num_hosts_remaining─┬─num_hosts_active─┐
│ clickhouse2 │ 9000 │      0 │       │                   1 │                0 │
│ clickhouse1 │ 9000 │      0 │       │                   0 │                0 │
└─────────────┴──────┴────────┴───────┴─────────────────────┴──────────────────┘

2 rows in set. Elapsed: 0.113 sec.

# Execute Insert From clickhouse1 node
clickhouse1 :) INSERT INTO Example_DB.product VALUES (now(), 765, 23);

INSERT INTO Example_DB.product FORMAT Values

Query id: 2685e1d1-6a8f-4e4f-a0cf-0ac110b987c6

Ok.

1 row in set. Elapsed: 0.011 sec.

# Select From clickhouse2 node
clickhouse2 :) select * from Example_DB.product;

SELECT *
FROM Example_DB.product

Query id: d6680636-1dec-472c-b90d-3731d27400ec

┌──────────created_at─┬─product_id─┬─category─┐
│ 2023-04-16 13:00:29 │        765 │       23 │
└─────────────────────┴────────────┴──────────┘

1 row in set. Elapsed: 0.001 sec.

clickhouse2 :)

Some more queries can be useful to verify cluster status using system tables.

clickhouse01 :) SELECT
    hostName(),
    database,
    name
FROM clusterAllReplicas(cluster_demo_ash, system.tables)
WHERE (database = 'Example_DB') AND (name = 'product')

Query id: a25221f0-d818-4897-b27f-dce27acbdc8a

┌─hostName()───┬─database───┬─name────┐
│ clickhouse01 │ Example_DB │ product │
└──────────────┴────────────┴─────────┘
┌─hostName()───┬─database───┬─name────┐
│ clickhouse02 │ Example_DB │ product │
└──────────────┴────────────┴─────────┘
┌─hostName()───┬─database───┬─name────┐
│ clickhouse03 │ Example_DB │ product │
└──────────────┴────────────┴─────────┘

3 rows in set. Elapsed: 0.193 sec.

clickhouse01 :) SELECT *
FROM system.zookeeper
WHERE path = '/clickhouse/tables/cluster_demo_ash/1/Example_DB/product'
FORMAT Vertical ;


clickhouse01 :) SELECT * FROM system.clusters WHERE cluster='cluster_demo_ash' FORMAT Vertical;

To dockerize your application with the ClickHouse cluster and Zookeeper nodes, Please refer GitHub

Conclusion

In conclusion, this blog post aimed to provide a comprehensive guide to building a simple two Node Database Cluster and one Node Zookeeper for data replication. It is important to note that for production use cases, a cluster with sharding and HA on Zookeeper may be necessary, and this requires careful consideration and planning. It is also essential to implement enhanced monitoring to keep a watchful eye on each component of the setup. Alternatively, to avoid these steps, you can utilize ChistaDATA DBaaS service for ClickHouse, which takes care of your infrastructure setup, maintenance, and management, allowing you to focus on your product and business. For more details, Visit Get started with ChistaData Guide