Contact Us

Home / Engineering Blog / Data Systems

Data SystemsEngineering notes

Working Effectively with Distributed Tables in ClickHouse

Gleb Kononenko and Alexey Dikov are developers at Digital League. A year ago, on a large project, the two of them and their colleagues started working with ClickHouse and immediately ran into a pile of problems — with little information available on how to get past them.

ClickHouse is a specialized, very fast database. The specificity lies in how data is stored and processed. Every table specifies an Engine — a storage engine that processes data asynchronously after loading. That processing can remove duplicates, sort data, replicate, and so on. The different engines are covered in more detail here.

The product is open source, ships with Russian-language documentation, and offers support options. Its growing popularity is no surprise.

The authors picked up experience the hard way, in practice, and are ready to share it — starting with how to create and use distributed tables efficiently.

A bit about the project:

  • Data loads every 15 minutes
  • Duplicate information keeps arriving
  • Data has to be retained for 5 years
  • On average, 150 million rows arrive per day (peak — up to 3 billion/day)
  • The database holds 685 billion rows: 35 TB compressed, 350 TB uncompressed

Contents

  • Cluster configuration
  • Creating tables
  • Distributing data across nodes
  • How the team moves data between tables
  • How the team resharded data after adding new nodes (shards)

Cluster configuration

First, a look at the test ClickHouse setup. The database runs on four virtual machines: clickhouse01-clickhouse4. The configuration describes three clusters: main, all-sharded and all-replicated. Their settings can be seen in the system.clusters system table:

select cluster, shard_num, replica_num from system.clusters;

all-replicated  1   1
all-replicated  1   2
all-replicated  1   3
all-replicated  1   4
all-sharded     1   1
all-sharded     2   1
all-sharded     3   1
all-sharded     4   1
main            1   1
main            1   2
main            2   1
main            2   2

The all-replicated cluster: shard_num is 1 everywhere, while replica_num takes values from 1 to 4. Tables created inside this cluster are therefore always replicated — the data on every node is completely identical.

This is how reference data, small entities, and finished marts are stored. It avoids unnecessary network transfers between nodes, and keeps the data highly available when one node fails.

The all-sharded cluster: shard_num takes values from 1 to 4, while replica_num is 1 everywhere. Tables in this cluster don't need replication but do need sharding (distribution) — each node holds its own instance of the data, and no other node has the same slice.

Such tables are useful for raw data, intermediate calculations, or any other case where high availability isn't required (if 1 of the 4 nodes goes down, only 75% of the total data volume stays available).

The main cluster: both shard_num and replica_num take values from 1 to 2. Distributed tables with replicas are created in this cluster. clickhouse01 and clickhouse03 act as the primary shards, while clickhouse02 and clickhouse04 are their replicas respectively.

Large tables that need data distribution go into this cluster, and they also need high availability (if one node fails, users still get the full data volume).

On top of that, macros have to be configured on each node to declare which shard and/or replica it is. They are expanded in the paths of the ZooKeeper coordinator service when tables are created, roughly like this:

<macros>
   <cluster>main</cluster>
       <shard>1</shard>
     <replica>1</replica>
</macros>

Creating tables

Since a load balancer is used, each request is sent round-robin to one of the nodes. In other words, when you send a command to ClickHouse, you don't know which node it will run on.

What happens if a table is created like this?

create table db.users (
   uid Int16,
   name String,
   age Int16
)
ENGINE = MergeTree
ORDER BY (uid);

The answer is simple — nothing good. The table gets created on one random node, and a later insert/select fails with a 75% probability of a "table does not exist" error, since the request will most likely be sent to a node that doesn't have it. The table has to be created manually on all nodes, or the on cluster option can be used, which automatically sends the command to every node in the cluster.

create table db.users on cluster 'cluster-name'...;

But that is actually not enough either. Table engines have to be specified correctly. The MergeTree engine from the example above is the most common one: it stores data sorted by the key, which makes very fast lookups possible. Read more in the documentation.

If the table is replicated — that is, a copy of it is written to several nodes — the engine subtype called ReplicatedMergeTree should be specified instead. In that case the engine parameters must include the table creation path for ZooKeeper.

For tables in the all-replicated cluster, the path is:

ENGINE =ReplicatedMergeTree('/clickhouse/tables/all-replicated/TABLE_NAME', '{replica}')

The braces hold a variable that is returned by the macro placed on each node — for example, {replica} returns the value 1 (see the macro example above). This way ZooKeeper builds unique paths on its side for distributing the data. For all-replicated, only replicas are used, without sharding. More on table replication is here.

For tables in the main cluster, the path is:

ENGINE =ReplicatedMergeTree('/clickhouse/tables/{shard}/TABLE_NAME, '{replica}')

The macro {shard} is substituted with the shard number in ZooKeeper, and {replica} with the replica number.

An example of creating a table in the main cluster:

create table db.users on cluster 'main'(
   uid Int16,
   name String,
   age Int16
)
ENGINE =ReplicatedMergeTree('/clickhouse/tables/{shard}/users', '{replica}')
ORDER BY (uid);

Distributing data across nodes

Most entities are stored in the main cluster. Data replicas provide fault tolerance, and sharding provides horizontal scaling.

Suppose a table has been created on all nodes of the main cluster. The insert command runs on some random node, and all the data stays on a single shard. The data is automatically replicated to the shard's replica if replication is configured. To spread the data across all shards, a table with the same structure, a different name, and the Distributed engine has to be created. Such a table holds no data — it is a kind of view that joins the tables from every shard of the cluster.

create table db.users_distributed on cluster 'main' as db.users;
ENGINE = Distributed('main', 'db', users, rand())

When data is inserted into this table, it is distributed evenly and appears on all the nodes, but at random.

That variant doesn't work for this setup, because when designing the storage the goal is to minimize data transfers over the network and keep all calculations local within a single node.

For that reason, distribution of data by a specific field or fields is preferred. One ClickHouse constraint: the distribution field must have the int data type. If the distribution has to be by a field of another type, hash functions that return an integer value can be used (murmurHash/cityHash/xxHash, for example).

Here is an example of creating a table with a distribution key built from a hash of a specific table field:

create table db.users_distributed on cluster 'main' as db.users;
ENGINE = Distributed('main', 'db', users, murmurHash3_32(name))

The data is now distributed by the value of

mod(murmurHash3_32(name), X), where X is the number of shards.

ClickHouse allows data to be distributed unevenly between shards. For example, if no data redistribution (resharding) is planned when new nodes are added, a larger share of the data can be sent to the new nodes. To do this, a weight can be set for the shard in its configuration:

<shard>
    <weight>10</weight>
    <replica>
        <host>clickhouse01</host>
        <port>9000</port>
    </replica>
</shard>
<shard>
    <weight>15</weight>
    <replica>
        <host>clickhouse03</host>
        <port>9000</port>
    </replica>
</shard>

In that case the data is distributed not by the number of shards but by their weight: mod(murmurHash3_32(name), X), where X equals the total weight. So 10/25 of the data goes to the first shard and 15/25 to the second (10 and 15 are the shard weights from the configuration above).

How the team moves data between tables

Moves within a single node

A feature of this project: data has to be moved between tables several times with some transformation along the way. The first attempt was the direct one — take the distributed tables and run a single command:

insert into db.target_distributed
select * from db.source_distributed;

But practice showed that this is extremely slow. It is far more efficient to run the data move locally inside each shard, connecting directly to the node in question:

insert into db.target
select * from db.source;

This way two operations are eliminated at once: the data transfer over the network and the re-distribution of data between nodes.

If there is no way to connect to the nodes — when a load balancer is used, for instance — a local move can be done through the remote function:

insert into function remote('clickhouse01', 'db.target')
select * from remote('clickhouse01 ', 'db.soure');

When using the remote function, either the node's ip address or its hostname can be used (if the nodes are on the same network).

Moving data between nodes

This option can be necessary when resharding data (when new nodes are added).

The simple variant is again from a distributed table into a distributed table. Or the data can be loaded in parallel from each shard into the target distributed table:

insert into db.target_distributed select * from db.source;

There is a more advanced option: pre-select slices of data, calculate which target shard they should end up in, and send them there directly.

insert into function remote('clickhouse02', 'db.target')
select * from remote('clickhouse01', 'db.soure')
where mod(distributed_field, X) = 1;

That is, data is loaded from the first shard into the second, selecting on the first shard the records where the remainder of the distribution field divided by X (the number of shards) equals the shard number minus one.

How the team resharded data after adding new nodes (shards)

The data volume keeps growing, and the system has to grow with it — ClickHouse allows both vertical and horizontal scaling. With horizontal scaling, tables in the all-replicated cluster are automatically replicated onto the new nodes. The remaining tables have to be resharded manually.

The problem was that after adding a shard, the condition changed:

mod(murmurHash3_32(name), X)

The number of shards became larger, so new data with a given distribution key can now land on the new shard (X in the formula changed!). That is unacceptable if the calculations are to stay local. There is only one way out — move the data using temporary tables.

But even here there turned out to be plenty of pitfalls. No long preamble — this is what the team did:

  1. Get the list of commands for creating the tables that need to be resharded:
with
    trim(TRAILING '_' from trim(TRAILING 'distributed' from name))
        as   table_name
select concat('show create ', database, '.', table_name ,';') from system.tables
where engine = 'Distributed';
  1. Run the selected commands one by one, take the table creation script and edit it. Add the '_tmp' postfix to the name, add the 'on cluster' option after the table name, and edit the path for ZooKeeper: the _tmp postfix is added to the table name there as well. Here is the result:
create table db.users_tmp on cluster 'main'(
   uid Int16,
   name String,
   age Int16
)
ENGINE =ReplicatedMergeTree('/clickhouse/tables/{shard}/users_tmp', '{replica}')
ORDER BY (uid);

Temporary tables are created for resharding the data.

On top of them, a distributed table is created:

create table db.users_tmp_distr on cluster 'main' as db.users_tmp;
ENGINE = Distributed('main', 'db', users_tmp, murmurHash3_32(name))
  1. Now an insert into the distributed table has to be launched from every shard. To speed this process up, the data was filtered and loaded straight into the needed shard. To do that, the records are selected where the remainder of the sharding field divided by the total number of shards in the cluster is equal to the shard number minus one. So the remainder is 0 for the first shard, 1 for the second shard, 2 for the third, and so on. At the end of the query, the setting is added, which writes the data to the shard we specify.

An example of launching the load for data that should end up on the first shard (a similar command with a changed where condition is run on every node):

insert into db.users_tmp_distr
select * from remote('clickhouse01', 'db.users')
where mod(murmurHash3_32(name), X) = 0   – X - new number of shards
settings insert_shard_id = 1;

insert into db.users_tmp_distr
select * from remote('clickhouse02', 'db.users')
where mod(murmurHash3_32(name), X) = 0   – X - new number of shards
settings insert_shard_id = 2;

This way all the new nodes are traversed with different parameters to move all the data into the new table. If there were 2 nodes and there are now 3, that means 6 insert commands have to be executed.

  1. Check that all the data is on its proper shards:
select distinct shardNum(),
                mod(murmurHash3_32(name),3)
from db.users_tmp_distr;
    1    0
    2    1
    3    2

No shard should repeat in the query result — that means all the data is distributed correctly.

  1. Swap the tables and delete the temporary ones:
exchange tables db.users and db.users_tmp on cluster 'main';
drop table db.users_tmp on cluster 'main' sync;
drop table db.users_tmp_distr on cluster 'main' sync;

_____________________

So, this article covered the different clusters in ClickHouse:

  • all-sharded — the fastest processing without replication, temporary data;
  • all-replicated — no network transfers, the distribution key doesn't matter, reference data;
  • main — shards and replication, a balance between availability and performance, the bulk of the tables.

A well-designed data distribution is the key to success. Later on you can use local loads and calculations, which speeds them up significantly. The team cut its data move time almost 6× compared with its first experiments. On top of that, it learned to redistribute data efficiently when new nodes are added.

← Back to all engineering notes