Contact Us

Home / Engineering Blog / Data Systems

Data SystemsEngineering notes

Home Sweet Home: The Inner Workings of ClickHouse, Part 2 — Replication

Pyotr is an engineer at Nixys. This article looks at the finer points of working with this columnar database and covers the replication process in detail.

First, a look at the terminology. It is basic, but ClickHouse has a few nuances worth spelling out upfront.

Replica

A plain copy of the data. ClickHouse always keeps at least one copy of your data, so the minimum number of replicas is one. An important detail: the original copy does not count as a replica, but ClickHouse code and documentation follow a convention where every instance is a replica.

Shard

A subset of the data, in which sharded data is split into non-overlapping sets. Say you have a table with a large volume of logs that physically does not fit on one server. You can split the table into 4 parts and place them on 4 different servers. This not only reduces the data volume on a single instance, but also increases system throughput and the number of I/O operations, since the operations run on all hosts.

ClickHouse always has at least one shard for your data. So if you do not spread the data across several servers, your data is stored in a single shard. The target server is chosen by the partitioning key, which is set when the distributed table is created.

Metadata

Information that describes other data, presented as structured reference records that help sort and identify the attributes of the information they describe.

Now we can move on to replication itself. Let's start with why it is needed in ClickHouse in the first place. Originally, in distributed systems, replication solved 3 core tasks:

  • High availability:

The idea is simple: when one instance is lost, the second keeps running, and traffic can be switched only to the healthy node. If a replica is unavailable, replication does not fail — the data is written to it once it is back in service.

  • Load scaling;

Just as straightforward: when the number of queries or the data volume grows, you bring another ClickHouse instance into the system and scale horizontally. Topology-wise, ClickHouse allows different architectures: you can use sharded-only tables or replication-only tables, as well as combined variants in which the shards are replicated.

  • Simplifying migration and updates;

Upgrading the ClickHouse version without downtime here is similar to other clustered systems: essentially you stop the data flow to the first instance, take it offline, update it, and bring it back into the cluster. Then repeat the operation on the second instance, and you get an upgraded cluster with no downtime at all.

However, as a project grows, ClickHouse replication turns from a luxury into a necessity. Why? It all comes down to table sizes: ClickHouse typically stores historical data, and that volume inevitably grows. As the business grows, the incoming data volume can easily reach hundreds of gigabytes a day. The last thing you want at that pace is to lose data in an outage. At a small scale, backups solve this problem, but they cover, at best, a day. Restoring from a backup takes time not only to load the backup into the instances, but also to restart all ETL processes so that the data created between the moment of the backup and the cluster failure is written again. In practice this takes a significant amount of time and usually blocks the processes of many teams. Replication solves this problem very effectively, especially geo-distributed replication — more on that below.

In MySQL, replicas receive data through the binary log — a log file where the master records its operations. The replica copies that log to itself and sequentially applies the operations from it.

The replication concept in ClickHouse is similar, but the database itself is not responsible for storing that log. Storing and coordinating replication actions in ClickHouse requires a distributed data store that can guarantee state consistency. ClickHouse uses one of two options: Zookeeper or ClickHouse Keeper.

Keeper does not store the table data itself, but it holds metadata that says which data should be stored in each replica. For example, if you insert 10 rows, ClickHouse creates a part, and a reference to that part is stored in Keeper. All ClickHouse instances know that the part exists in the cluster and needs to be found in the local file system. If it is not local, the instance must contact the other instances to copy it. Metadata correctness and Keeper availability itself are critical for replication to work correctly: if you lose your Keeper, all replicated tables go into read-only mode until Keeper returns to working order. ClickHouse now has a SystemRestoreReplica function that allows replication to be restored from local data, but this option should be used only as a last resort. Besides storing metadata, ClickHouse also uses Keeper to coordinate the cluster.

This article focuses mainly on working with ZooKeeper, since that store is used on most of the team's projects, but first a few words about ClickHouse Keeper.

ClickHouse Keeper is a replacement for ZooKeeper that supports its own extension optimized for working with ClickHouse. ClickHouse Keeper can be maintained in two different setups: as a standalone installation on separate servers, or as part of ClickHouse in the form of Process Keeper — part of the ClickHouse clickhouse-server binary. Although the second option looks simpler and cheaper (since Keeper is effectively built into ClickHouse), running ClickHouse Keeper as a standalone installation is strongly recommended, because it improves the fault tolerance of the system as a whole.

If you want to migrate from ZooKeeper to ClickHouse Keeper, keep 2 nuances in mind. First, ClickHouse Keeper snapshots and logs have a format incompatible with ZooKeeper, although you can convert ZooKeeper data into a ClickHouse Keeper snapshot with clickhouse-keeper-converter. Second, the ClickHouse Keeper inter-server protocol is incompatible with ZooKeeper, so creating a mixed ZooKeeper + ClickHouse Keeper cluster is impossible.

Now back to Zookeeper, viewed in the context of a ClickHouse cluster. Inside Zookeeper, all cluster data is stored in the /clickhouse directory, which contains the following:

  • ddl — all DDL queries that need to be replicated and applied to the replicas.
  • metadata — stores part of the metadata of a replicated table, for example the primary key, partition key, and so on;
  • columns — stores information about the fields of the replicated table;
  • block_numbers — stores all partitions of the table;
  • leader_election — used to elect a master among replicas. When a request comes in, a leading replica is selected first, and then another replica is synchronized from it. Several replicas can be leaders at the same time;
  • log — a directory that serves as a task queue for storing operations on the table replica;
  • blocks — stores hash information about data blocks written to the table over a certain period of time, for deduplication. The format of the child nodes is partition_hash_hash;
  • mutation — a directory that serves as a task queue for storing mutation operations on the replicated table;

  • replicas/node_hostname — stores information about the replicas:

  • is_active — a flag indicating whether the instance is active; if the server fails, the node will not exist and will be added again after recovery;

  • mutation_pointer — stores the next task in the mutation queue to be executed;
  • log_pointer — stores the next task in the log queue to be executed;
  • is_lost — a flag indicating whether the replica is stale, based on whether the log_pointer has been updated; 0 means normal, -1 means stale, and 1 means recovery in progress;
  • metadata — stores information about the table metadata, same as metadata above;
  • columns — stores information about the table columns;
  • parts — stores information about the table parts; each information block contains checksums and column information;
  • queue — a temporary processing queue

Now let's look at a standard implementation of a ClickHouse cluster. This scheme uses 5 servers: two machines for the master-master ClickHouse instances and three machines for the ZooKeeper cluster:

The following ports are used in this scheme:

  • 9000 — the standard ClickHouse port;
  • 9009 — the standard port for exchanging information between replicated tables;
  • 2181 — the standard port for communicating with ZooKeeper;

Now take the same scheme and make it geo-distributed:

The main ClickHouse tier is in one data center, and the backup tier is in another data center and region. No writes go to the backup tier, because the network latency would be too noticeable in that case. However, if the main tier fails completely, it is enough to switch the writers to the backup cluster, which is self-sufficient and has the same data in the databases and the same metadata in Keeper (except what did not arrive because of the network) as the main cluster.

With the main topology covered, let's look at the replication implementation itself in ClickHouse in more detail.

From an implementation point of view, replication in ClickHouse is an extension on top of the MergeTree engine family. That is, when we talk about replication, we mean the normal behavior of a particular MergeTree implementation that can be replicated. Put simply, you add Replicated to the name of the MergeTree engine and get its replicated version.

ClickHouse replication works on master-master tables without any of the locking found in OLTP databases. This is achieved through the Shared Nothing architecture, which distributes all queries across individual cores. When you insert something on an instance, it is replicated asynchronously to the other replicas. By default, an INSERT query waits for write confirmation from only one replica. So if the data was written successfully to only one replica and the server with that replica fails, the written data will be lost. To prevent this situation, you can enable write confirmation from several replicas using the insert_quorum setting, but this negatively affects performance. Also, because of the asynchrony, data appears on the server with a small delay — as long as it takes to transfer a block of compressed data over the network — forming the so-called replication lag.

Within MergeTree you insert parts, and ClickHouse performs their insertion and merges them with other parts in the background; the number of threads for background tasks can be set with the background_schedule_pool_size setting. How the merge should be performed is the responsibility of the leader — one of the tables that ClickHouse picks as the responsible one, which drives background merging through the Keeper cluster and reports it to the replicas. Thus, unlike OLTP databases, in ClickHouse you should not worry about the leadership status of a replica. DDL and TRUNCATE operations are also coordinated if applied with ON CLUSTER. Thus, almost all insert or table modification operations will be replicated, except the following:

  • CREATE table;
  • DROP table;
  • DETACH table;
  • ATTACH table;
  • ALTER TABLE FREEZE;
  • ALTER TABLE MOVE TO DISK;
  • ALTER TABLE FETCH;

The operations above are local to the instance and must be performed on all nodes of the cluster separately.

If, during operation, you need to switch from a non-replicable MergeTree table to its Replicated version, there are two main options:

1) Manual: create a replicated table with the same engine and structure, then attach the partitions one by one with ATTACH. This option works on all ClickHouse versions.

2) Automatic: you can add the replication flag to the table directory, and ClickHouse automatically converts the table to a replicated one. This feature was introduced in version 24.2.

Everything described above suggests that replication in ClickHouse is a reliable construct with no points of failure. And in general that is exactly how it is: the team has found that ClickHouse replication is indeed reliable and almost never breaks. Almost…

Replication failure

The team ran into a situation like this once, but that did not make it any more pleasant.

One fine winter morning, the monitoring system notified the team about replication lag. The logs showed a timeout error and a memory limit violation:

DB::Exception: Received from <Ip>. DB::Exception: Memory limit (total) exceeded: would use 113.01 GiB (attempt to allocate chunk of 5240030 bytes), maximum: 113.01 GiB

It turned out to be a migration with too large a volume of operations, as a result of which it hung after processing the query. The query could not be replicated, and replication lag occurred. That was not a big problem: once the migration completed, replication would have gone through successfully. What was a problem was the restart of the instance running that migration, caused by a hardware failure of the node. That led to part of the tables going into Read Only mode. The system tried to establish a connection to ZooKeeper, which was available, but an exception was raised due to an error while interacting with it. Most likely this was because the heavy query had not completed, which prevented the information from being reconciled with ZooKeeper.

To solve this problem, the team terminated these processes, after which ClickHouse was restarted again on the second instance. After that, the tables left Read Only mode, but 3 problems appeared in replication: the tables contained more than 300 parts_to_check.

SELECT

    database,

    table,

    is_leader,

    is_readonly,

    is_session_expired,

    future_parts,

    parts_to_check,

    columns_version,

    queue_size,

    inserts_in_queue,

    merges_in_queue,

    log_max_index,

    log_pointer,

    total_replicas,

    active_replicas

FROM system.replicas

WHERE is_readonly OR is_session_expired OR (future_parts > 20) OR (parts_to_check > 10) OR (queue_size > 20) OR (inserts_in_queue > 10) OR ((log_max_index - log_pointer) > 10) OR (total_replicas < 2) OR (active_replicas < total_replicas)

┌─database─┬─table────────────────┬─is_leader─┬─is_readonly─┬─is_session_expired─┬─future_parts─┬─parts_to_check─┬─columns_version─┬─queue_size─┬─inserts_in_queue─┬─merges_in_queue─┬─log_max_index─┬─log_pointer─┬─total_replicas─┬─active_replicas─┐

│ <db_name>       │ <table_name1> │         0 │           0 │                  0 │            0 │              0 │              -1 │        133 │              129 │               4 │      26974009 │    26974010 │              2 │               2 │

│ <db_name>       │ <table_name2> │         0 │           0 │                  0 │            5 │            333 │              -1 │         10 │                5 │               5 │      73264917 │    73264918 │              2 │               2 │

│ <db_name>       │ <table_name3> │         0 │           0 │                  0 │            0 │              0 │              -1 │        458 │              458 │               0 │       9938773 │     9938774 │              2 │               2 │

└──────────┴─────────────────────────┴───────────┴─────────────┴────────────────────┴──────────────┴────────────────┴─────────────────┴────────────┴────────

──────────┴─────────────────┴─

When a ClickHouse instance starts and establishes a new connection with Keeper, it checks the number and size of all files to find out whether the data set in the local file system matches the expected data set, comparing the actual size with the data about the expected sizes from Keeper.

  • If there are minor discrepancies, the system eliminates them by synchronizing data with the replicas.
  • If a SELECT query reads files that match in size but have bytes changed somewhere in the middle compared to the information in Keeper, the query will receive an exception about a mismatched checksum or compressed block size. Parts will be added to the check queue (parts_to_check) and, if necessary, downloaded from the replicas.
  • And if the local data set differs too much from what is expected from Keeper, ClickHouse's safety mechanism kicks in: the server will log it and refuse to start. In this case the ClickHouse instances did not fail, but part of the tables was corrupted:
<Error> xr.rep_site_recommendation (ReplicatedMergeTreePartCheckThread): No replica has part covering 1664323200_13512_13512_0 and a merge is impossible: we didn't find smaller parts with either the same min block or the same max block.

Starting with ClickHouse version 21.7, a powerful tool appeared: SYSTEM RESTORE REPLICA, which allows restoring a read-only replica if the data is present but the ZooKeeper metadata is lost. But the ClickHouse version in this case was older, so it could not be used.

In this case there is a manual solution: you need to create a restore flag file in the Keeper directory, which removes the existing information about the instance from Keeper and replicates all the data. To start the restore, the team created an empty file in the Keeper directory like this:

create /clickhouse/tables/<node_name>/<table_name>/replicas/<replica_name>/flags/force_restore_data

After that, ClickHouse was restarted. This solved the problem, and soon replication recovered.

In summary:

  • Keeper should be located on separate servers, preferably on SSDs, in 3, 5 or 7 nodes;
  • Do not clear Keeper manually;
  • Back up the information from Keeper;
  • Regularly update the ClickHouse version so you can use all the current tools;
  • Optimize your queries.

That's it for this article. Today we looked at the main points related to replication in ClickHouse, and briefly examined the process of analyzing and restoring replication after a failure.

← Back to all engineering notes