In April, Avito engineers planned an online gathering with ClickHouse's lead developer Alexey Milovidov and Kirill Shvakov, a Golang developer from Integros. They discussed how we use the database management system and the challenges we face.
Based on the meeting, we compiled an article with experts' answers to our and viewers' questions about backups, data resharding, external dictionaries, the Golang driver, and ClickHouse version updates. It may be useful for developers who are already actively working with Yandex's DBMS and are interested in its present and future. By default, the responses are from Alexey Milovidov, unless otherwise noted.
Caution, there's a lot of text under the cut. We hope the content with the questions helps you navigate.
Content
If you don't want to read the text, you can watch the recording of the gatherings. . The timestamps are in the first comment under the video.
ClickHouse is constantly updating, but our data is not. What should we do about this?
ClickHouse is constantly updating, but our data, which has been processed with optimize final, is not being updated and remains in backup.
Suppose we encountered some kind of problem, and data was lost. We decided to recover, and it turned out that old partitions saved on backup servers greatly diverge from the currently used version of ClickHouse. What should we do in such a situation, and is it possible?
A situation where you restored data from a backup in an old format, and it does not connect to the new version, is impossible. We ensure that the data format in ClickHouse remains backward compatible. This is much more important than backward compatibility in functionality if the behavior of some rarely used function has changed. The data stored on disk should always be readable by the new version of ClickHouse. This is the law.
What are the best practices currently for backing up data from ClickHouse?
How to make backups considering that we have optimize final operations, a massive database in terabytes, and data that has been updated, let's say, over the last three days, with no further procedures occurring with it?
We can come up with our own solution and write in Bash: collect these backups in this way. Perhaps we don't need to invent anything, and the wheel has already been invented?
First of all, regarding best practices. My colleagues always advise, in response to questions about backups, to mention the service 'Yandex.Cloud', where this task has already been solved. So use it if you have that opportunity.
There is no complete solution, fully integrated into ClickHouse, for backups. There are some templates that can be used. To achieve a complete solution, you will either have to work a bit manually or create wrappers in the form of scripts.
I will start with the simplest solutions and end with the most sophisticated, depending on the volume of data and the size of the cluster. The larger the cluster, the more complex the solution becomes.
If a data table occupies only a few gigabytes, a backup can be made as follows:
- Save the table definitions, that is, the metadata — show create table.
- Make a dump using the ClickHouse client — select * from table to a file. By default, you will receive a file in TabSeparated format. If you want something more efficient — you can use the Native format.
If the volume of data is larger, the backup will take more time and a lot of space. This is called a logical backup; it is not tied to the ClickHouse data format. If it exists, in extreme cases, you can take the backup and load it into MySQL for recovery.
For more advanced cases, ClickHouse has the ability to create snapshots of partitions in the local file system. This feature is available as a query alter table freeze partition. Or simply alter table freeze — this is a snapshot of the entire table.
The snapshot will be created consistently for one table on one shard; that is, it is impossible to create a consistent snapshot of the entire cluster in this way. But for most tasks, this is not necessary, and it is sufficient to execute the query on each shard and obtain a consistent snapshot. It is created as hard links, so it does not take up additional space. You then copy this snapshot to the backup server or to the storage that you use for backups.
Restoring such a backup is quite easy. First, you create tables based on the existing table definitions. Then you copy the saved partition snapshots into Directory-Detached for the respective tables and execute the query attach partition. This solution is quite suitable for the most serious volumes of data.
Sometimes something even more impressive is required — in cases where you have tens or even hundreds of terabytes on each server and hundreds of servers. There's a solution I noticed from colleagues at "Yandex.Metrica". I wouldn't recommend it for everyone — read it and decide for yourself if it's suitable.
First, you need to create several servers with large disk shelves. Next, set up several ClickHouse servers on these servers and configure them to work as another replica for the same shards. Then use a filesystem or some tool that allows you to create snapshots on these servers. There are two options here. The first option is LVM snapshots; the second option is ZFS on Linux.
After that, you need to create a snapshot every day, which will take up some space. Naturally, if the data changes, the volume of space will increase over time. This snapshot can be retrieved at any moment to restore the data, such a strange solution. Additionally, you need to limit these replicas in the config to prevent them from trying to become leaders.
Will it be possible to organize controlled lag for replicas in the clusters?
This year, are you planning to create waves in ClickHouse? Will it be possible to organize controlled lag of replicas within them? We would like to use it to protect ourselves from negative scenarios involving alternates and other changes.
Is it possible to perform any rollbacks for alternates? For example, in an existing wave, can we specify to apply changes only up to this point and stop applying changes from this moment forward?
If a team has come to our cluster and broken it, do we have a conditional replica with a one-hour lag where we can say, let’s use this one for now, but do not apply the last ten minutes of changes to it?
First, let’s talk about controlled lag of replicas. There was a request from users, and we created an issue on GitHub asking: "If anyone needs this, please like it, give it a heart." Nobody liked it, and the issue was closed. Nevertheless, it is already possible to get this functionality by configuring ClickHouse, but only starting from version 20.3.
ClickHouse constantly performs background data merging—merge. When a merge is completed, a certain set of data parts is replaced by a larger piece. Meanwhile, the earlier data parts remain on the disk for some time.
Firstly, they continue to be stored until there are select queries that use them, to ensure non-blocking operation. Select queries can read from the old parts without issues.
Secondly, there is also a time threshold—old data parts sit on the disk for eight minutes. This eight minutes can be adjusted and even turned into a whole day. This will cost disk space: depending on the data stream, it is possible that in the last day the data will not just double, but could become five times larger. However, you can stop the ClickHouse server during a serious problem and deal with everything.
Now the question arises, how does this protect against alters? Here it is worth looking deeper because, in older versions of ClickHouse, alters worked in such a way that they simply directly changed the parts. There is a data part with some files, and we do, for example, alter drop column. In this case, that column is physically removed from all parts.
But starting from version 20.3, the alter mechanism has been completely changed, and now data parts are always immutable. They do not change at all—alters now work similarly to merges. Instead of changing a part in place, we create a new one. In the new part, files that haven’t changed become hard links, and if we delete a column, it will simply not be present in the new part. The old part will be deleted by default after eight minutes, and here you can tweak the settings mentioned above.
The same applies to mutation-type alters. When you do alter delete or alter update, it does not modify the part, but creates a new one. And then removes the old one.
What should we do if the table structure has changed?
How to restore a backup that was made with an old schema? And a second question about the case with snapshots and filesystem tools. Is Btrfs suitable here instead of ZFS on Linux LVM?
If you are doing attach partition If you have partitions with a different structure, ClickHouse will inform you that it's not allowed. The solution is as follows. First, create a temporary MergeTree table with the old structure, attach the data using attach, and then perform an alter query. After that, you can either copy or move this data and attach it again, or use the query. alter table move partition.
Now, the second question is whether Btrfs can be used. To start, if you have LVM, then LVM snapshots are sufficient, and the file system can be ext4, which doesn't matter. With Btrfs, it all depends on your experience with its operation. It is a mature file system, but there are still some concerns about how it will perform in practice in a specific scenario. I wouldn't recommend using it if you don't have Btrfs in production.
What are the best practices currently for resharding data?
The question about re-sharding is complex and multifaceted. There are several ways to answer this. One approach is to say that ClickHouse does not have built-in re-sharding capabilities. However, I fear that this answer won't satisfy anyone. Thus, we can approach it from another side and state that there are many ways to re-shard data in ClickHouse.
If you run out of space on the cluster or it can't handle the load, you add new servers. But these servers are empty by default, with no data on them, and no load is applied. You need to redistribute data so that it is evenly spread across the newly expanded cluster.
The first way to achieve this is to copy some partitions to the new servers using the query. alter table fetch partitionFor example, if you had partitions by months, you would take the first month of 2017 and copy it to a new server, then copy the third month to another new server. You continue this until it becomes more or less evenly distributed.
Data can only be moved for those partitions that do not change during writes. For fresh partitions, you will need to disable writes, because their transfer is not atomic. Otherwise, you will end up with duplicates or gaps in the data. Nevertheless, this method is practical and works quite effectively. Ready compressed partitions are sent over the network, meaning the data is not re-compressed or re-encoded.
This method has one drawback, which depends on the sharding scheme. It matters what you planned for this sharding scheme and what your sharding key was. In your case, for metrics, the sharding key is a hash of the path. When you execute a select on the Distributed table, it queries all the shards of the cluster and retrieves the data from there.
This means that, in reality, it doesn't matter to you which data ended up on which shard. The main point is that data for the same path resides on one shard, but which one is not critical. In this case, transferring pre-prepared partitions works perfectly because during select queries, whether before or after reshaping, the scheme of values doesn't matter—you will receive complete data.
However, there are more complex cases. If at the application logic level you rely on a special sharding scheme, where this client is located on a certain shard, and the query can be sent directly there rather than to the Distributed table. Or you are using a relatively new version of ClickHouse and have enabled the setting optimize skip unused shards. In this case, during a select query, the expression in the where section will be analyzed, and it will be determined which shards need to be accessed according to the sharding scheme. This works provided that the data is laid out according to this sharding scheme. If you have reshuffled them manually, the alignment may change.
So, this is method number one. I am waiting for your response, whether this method works for you or if we should move on.
Vladimir Kolobayev, lead system administrator at Avito: Alexey, the method you mentioned doesn't work very well when we need to evenly distribute the load, including for reading. We can take a monthly partition and move the previous month to another node, but when the request for that data comes in, we will be loading only it. Ideally, we want to load the entire cluster, as otherwise, for some time all read load will be handled by just two shards.
Alexey Milovidov: The response here is strange — yes, it's bad, but it might work. I'll explain how exactly. You should look at the load scenario that comes with your data. If this is monitoring data, then it's almost certain that the vast majority of requests are for fresh data.
You have set up new servers, migrated old partitions, but also changed how fresh data is recorded. Fresh data will be spread across the entire cluster. Thus, in just five minutes, requests for the last five minutes will evenly load the cluster; after a day, requests for the past day will evenly load the cluster. Unfortunately, requests for the previous month will only go to a part of the cluster's servers.
However, often you won't have requests specifically for February 2019. Most likely, if requests are going to 2019, they will be for all of 2019 — for a large time span, not for some small range. Such requests can also evenly load the cluster. But in general, your remark is entirely correct that this is such an ad hoc solution, which does not spread the data completely evenly.
I have a few more points to address the question. One of them is about how to originally design the sharding scheme so that there is less pain from resharding. This is not always possible.
For example, you have monitoring data. Monitoring data grows for three reasons. First — the accumulation of historical data. Second — an increase in traffic. And third — a growing number of items that fall under monitoring. New microservices and metrics that need to be recorded are emerging.
It is possible that the largest growth is related to the third reason — the increased use of monitoring. In this case, it's worth looking at the nature of the load, what the main select requests are. The main select requests will most likely be based on a certain subset of metrics.
For example, the CPU usage on certain servers by a particular service. It turns out that there is a subset of keys through which you retrieve this data. And the request for this data is likely quite simple and executed in tens of milliseconds. It is used for monitoring services, for dashboards. I hope I am understanding this correctly.
Vladimir Kolobaev: The thing is, we often refer to historical data, as we compare the current situation to the historical one in real time. It's important for us to have quick access to a large volume of data, and ClickHouse handles this excellently.
You are absolutely right, the majority of read requests we experience are from the last day, just like any monitoring system. However, historical data also has a significant load. This mainly comes from the alerting system, which queries ClickHouse every thirty seconds, asking, 'Give me the data for the last six weeks. Now, generate a moving average from them, and let’s compare the current value with the historical one.'
I would like to mention that we have a small table for such very fresh requests, where we store data for only two days, and most queries are directed there. We only send larger historical requests to the big sharded table.
Alexey Milovidov: Unfortunately, for your scenario, this is poorly applicable, but I will describe two bad and complex sharding schemes that should not be used, but are used in my friends' service.
There is a main cluster with events from 'Yandex.Metrics'. Events include page views, clicks, and transitions. Most queries are directed at a specific website. You open the 'Yandex.Metrics' service, you have a website — avito.ru, go to the report, and the request is made for your site.
But there are also other queries — analytical and global, made by internal analysts. Just to note, internal analysts only query the 'Yandex' services. However, even 'Yandex' services account for a significant share of all data. These queries are not for specific counters but for broader filtering.
How can we organize the data in such a way that it works efficiently for both individual counters and global queries? The complexity arises from the fact that the number of requests in ClickHouse on the "Metrics" cluster is several thousand per second. Furthermore, non-trivial requests, for example, several thousand per second, cannot be handled by a single ClickHouse server.
The size of the cluster is a bit over six hundred servers. If we simply apply a Distributed table over this cluster and send several thousand requests there, it would actually be worse than sending them to a single server. On the other hand, we can discard the option where data is evenly distributed, and we just query from all servers.
There is a diametrically opposite option. Imagine if we shard the data by websites, and the request for one website goes to one shard. Now the cluster can handle ten thousand requests per second, but on one shard, a particular request might be too slow. It won’t scale in terms of throughput. Especially if it's the website avito.ru. I won't reveal a secret by saying that Avito is one of the most visited websites in the Russian internet. Processing it on one shard would be madness.
Therefore, the sharding scheme is structured in a more sophisticated way. The entire cluster is divided into a certain number of clusters which we call layers. Inside each cluster, there are from a dozen to several dozen shards. In total, there are thirty-nine such clusters.
How does all this scale? The number of clusters remains unchanged — it has been thirty-nine for several years and continues to be. However, we gradually increase the number of shards within each of them as we accumulate data. The overall sharding scheme is such that the division into these clusters is done by websites, and to determine which website is on which cluster, a separate metadata database in MySQL is used. One website corresponds to one cluster. Inside it, sharding is done by visitor IDs.
When recording, we split them based on the remainder of the visitor identifier. However, when adding a new shard, the sharding scheme changes; we continue to split, but based on the remainder of the division by a different number. This means that one visitor is actually located on several servers, and you cannot rely on that. This is done solely to ensure that the data compresses better. For queries, we go to a Distributed table, which looks at the cluster and reaches out to dozens of servers. It's a frustrating scheme.
But my story would be incomplete if I didn't mention that we abandoned this scheme. In the new scheme, we've changed everything and copied all the data using clickhouse-copier.
In the new scheme, all sites are divided into two categories - large and small. I don't know how the threshold was determined, but as a result, large sites are recorded on one cluster, where there are 120 shards with three replicas each - that is, 360 servers. The sharding scheme is such that any request goes directly to all shards. If you open any report page for avito.ru in 'Yandex.Metrica' right now, the request will go to 120 servers. There are very few large sites in the Russian segment of the internet. Consequently, the requests total not a thousand per second, but even less than a hundred. The Distributed table easily handles this, with each of them processed by 120 servers.
The second cluster is for small sites. Here, the sharding scheme is based on the site identifier, and each request goes to exactly one shard.
ClickHouse has a utility called clickhouse-copier. Can you tell us about it?
I will say right away that this solution is bulkier and somewhat less efficient. The advantage is that it completely disperses the data according to the scheme you specify. However, the disadvantage of the utility is that it doesn’t perform resharding at all. It copies data from one cluster scheme to another.
This means that for it to work, you need to have two clusters. They can be located on the same servers, but nevertheless, the data will not be moved incrementally; they will be copied.
For example, there were four servers, and now there are eight. You create a new Distributed table on all servers, set up new local tables, and run clickhouse-copier, specifying the operational schema so that it reads from there, accepts the new sharding schema, and transfers the data there. You'll need one and a half times more space on the old servers than what you currently have, because the old data must remain there, and on top of that, half of this old data will arrive. If you had thought in advance that the data needed to be resharded and there is space available, then this method would be suitable.
How is clickhouse-copier structured internally? It breaks down all the work into a set of tasks for processing one partition of one table on one shard. All these tasks can run simultaneously, and clickhouse-copier can be launched on different machines in multiple instances, but what it does for one partition is nothing more than an insert select. Data is read, decompressed, redistributed, then compressed again, written somewhere, and resorted. This is a heavier solution.
You had a pilot project called resharding. What happened to it?
You had a pilot project back in 2017 called resharding. There's even an option in ClickHouse. I understand it didn't take off. Can you explain why that happened? It seems quite relevant.
The whole problem is that when there is a need to reshuffle data on-site, a rather complex synchronization is required to do it atomically. When we started to look at how this synchronization is structured, it became clear that there are fundamental problems. And these fundamental problems are not just theoretical but immediately started to manifest in practice in a way that can be explained very simply—nothing works.
Is it possible to merge all data parts together before moving to slower disks?
A question about TTL with the move to slow disk option in the context of merges. Is there a way, other than cron, to merge all parts into one before moving to the slow disks?
The answer to the question of whether there's a way to automatically merge all pieces into one before their transfer is no. I don't think it's necessary. You can choose not to merge all parts into one, but simply count on them being transferred to the slow disks automatically.
We have two criteria for migration rules. The first is based on how full the storage is. If there is less than a certain percentage of free space at the current storage level, we select one chunk and move it to slower storage. Not slower per se, but to the next one—depending on how you configure it.
The second criterion is based on size. This is about moving large chunks. You can adjust the threshold for free space on the fast disk, and the data will be migrated automatically.
How can we migrate to new versions of ClickHouse if there’s no opportunity to check compatibility in advance?
This topic is regularly discussed , taking into account different versions; however, how safe is it to upgrade from version 19.11 to 19.16 and, for instance, from 19.16 to 20.3? What is the best way to migrate to new versions without the ability to test compatibility in a sandbox beforehand?
There are a few 'golden' rules. The first is— . It is extensive, but there are separate sections about backward-incompatible changes. These points should not be seen as red flags. Usually, they are minor incompatibilities related to some fringe functionality that you likely do not use.
Second—if you do not have the opportunity to check compatibility in a sandbox and want to upgrade directly in production, the recommendation is this—do not do it. First, create a sandbox and check. If there is no testing environment, then you are most likely not a very large company, meaning you can copy part of the data to your laptop and verify that everything works correctly there. You can even spin up several replicas locally on your machine. Alternatively, you can set up a new version nearby and load some of the data there—essentially creating an improvised testing environment.
Another rule is not to upgrade within a week after a version release due to bug hunting in production and subsequent quick fixes. Let's clarify ClickHouse versioning to avoid confusion.
There is version 20.3.4. The number 20 indicates the year of release — 2020. From the perspective of what's inside, it doesn't really matter, so we won't pay attention to it. Next is 20.3. We increase the second digit — in this case, 3 — every time we release a version with some new functionality. If we want to add a feature to ClickHouse, we are required to increase this number. That means in version 20.4 ClickHouse will work even better. The third digit — 20.3.4. Here, 4 is the number of patch releases, in which we didn't add new features but fixed some bugs. And 4 means we did this four times.
Don't think of it as something terrible. Usually, a user can install the latest version, and it will work without any uptime issues for a year. But imagine that in some function for processing bitmaps, which was added by our Chinese colleagues, the server crashes when incorrect arguments are passed. We have to fix this. We will release a new patch version, and ClickHouse will become more stable.
If you have ClickHouse running in production, and a new version of ClickHouse with additional features comes out — for example, 20.4.1 — don’t rush to deploy it in production on the first day. Why is it needed at all? If you haven't started using ClickHouse yet, you can install it, and most likely, everything will go well. But if ClickHouse is already running reliably, keep an eye on patches and updates — what problems we are fixing.
Kirill Shvakov: I want to add a bit about test environments. Everyone is very afraid of test environments and somehow believes that if you have a very large ClickHouse cluster, the test environment should be at least as large or at least ten times smaller. This is not the case at all.
I can speak from my own experience. I have a project that uses ClickHouse. Our test environment for it is a small virtual machine in Hetzner for twenty euros, where absolutely everything is deployed. To manage this, we have full automation in Ansible, and therefore it really doesn’t matter whether it’s on physical servers or simply deployed in virtual machines.
What can be done? It would be helpful to provide an example in the ClickHouse documentation on how to set up a small cluster—using Docker, LXC, or possibly creating an Ansible playbook, as different people have different deployment methods. This would simplify a lot. When you can deploy a cluster in five minutes, it's much easier to figure things out. It's much more convenient this way because rolling out a production version that you haven't tested is a path to nowhere. Sometimes it works, and sometimes it doesn't. So hoping for success is not a good strategy.
Maxim Kotyakov, Senior Backend Engineer at Avito: I'll add a bit about the testing environments from the perspective of issues faced by large companies. We have a full-fledged acceptance cluster for ClickHouse, which is an exact copy of our production environment in terms of data schemas and settings. This cluster is deployed in rather shabby containers with minimal resources. We write a certain percentage of production data there, thanks to the ability to replicate streams in Kafka. Everything there is synchronized and scaled—both in terms of capacity and flow, and theoretically, under equal conditions, it should behave like production in terms of metrics. Everything potentially explosive first goes to this stand and sits there for a few days until it's ready. But naturally, this solution is expensive, heavy, and incurs non-zero maintenance costs.
Alexey Milovidov: Let me tell you about the testing environment of our friends at Yandex.Metrics. One cluster had over 600 servers, another had 360, and there is a third one and several more clusters. The testing environment for one of them is simply two shards with two replicas in each. Why two shards? So that there isn't just one. And also to have replicas. Just a minimal amount that one can afford.
This testing environment allows checking the functionality of queries and whether something major has broken. However, problems often arise of a completely different nature, where everything works but there are some slight changes under load.
For example, we decided to install a new version of ClickHouse. It has been deployed to the testing environment, automated tests in Yandex.Metrics have been conducted that compare data from the old version to the new one, running the entire pipeline. And, naturally, our CI tests showed green results. Otherwise, we wouldn't have even proposed this version.
Everything is perfect. We are rolling it out into production. I receive a message that the load on the graphs has increased several times. We roll back the version. I look at the graph and see: the load has indeed increased several times during the rollout, and it decreased back when we rolled it out. Then we started to roll back the version. And the load increased just as much and subsequently fell back as well. So the conclusion is this — the load increased due to the deployment, which is nothing surprising.
After that, it was difficult to convince my colleagues to install the new version anyway. I said, "Everything is fine, roll it out. Keep your fingers crossed, everything will work. Right now, the load has increased on the graphs, but it's all good. Hang in there." In general, we did that, and everyone — the version is live in production. But similar problems arise almost with every deployment.
Kill query is supposed to kill queries but it doesn’t seem to work. Why is that?
A user came to me, some kind of analyst, and created a query that brought down my ClickHouse cluster. Some node or the entire cluster — depending on which replica or shard the query hit. I see that all CPU resources on this server are at their peak, everything is red. At the same time, ClickHouse is still responding to queries. And I write: "Please show me the process list, which query caused this madness."
I find this query and write kill. And I see that nothing happens. My server is at peak load, ClickHouse continues to give me some commands, showing that the server is alive, and everything is great. But I have degradation in all the user queries, degradation in writing to ClickHouse begins, and my kill query is not executing. Why? I thought that kill query should terminate queries, but that’s not happening.
Now there will be a rather strange answer. The fact is that kill query does not terminate queries.
Kill query sets a small flag called "I want this query to be killed." The query itself, while processing each block, checks this flag. If it is set, the query stops working. It turns out that no one kills the query; it has to check everything and stop by itself. And this should work in all cases when the query is in the process of handling data blocks. It will process the next data block, check the flag, and stop.
This does not work in cases where the request is blocked for some operation. However, it is likely that this is not your situation, as you mentioned it uses a lot of server resources. It may not work in the case of external sorting and a few other details. But overall, this should not happen; it's a bug. The only advice I can give is to update ClickHouse.
How to calculate response time under read load?
There is a table that stores aggregates by item — various counters. The number of rows is about one hundred million. Can we expect predictable response times if we pump in 1K RPS across 1K items?
Judging by the context, this is about read load, because there are no issues with writes — you can insert a thousand, a hundred thousand, or even several million rows.
Read requests can vary widely. In a select 1, ClickHouse can perform around tens of thousands of queries per second, so even queries by a single key will require some resources. These point queries will be more complex than in key-value databases because for each read, you need to read a block of data by index. The index addresses not every record, but each range. This means you will have to read the entire range — which is 8192 rows by default. Additionally, you will need to decompress a data block from 64 Kb to 1 Mb. Typically, such point queries take a few milliseconds. But this is the simplest case.
Let’s try some simple arithmetic. If you multiply a few milliseconds by a thousand, you will get several seconds. It seems like you can't handle a thousand requests per second, but in reality, you can because we have multiple CPU cores. So, in principle, 1000 RPS is something ClickHouse can sometimes manage, but only for short, point queries.
If you need to scale a ClickHouse cluster for the number of simple queries, I recommend the simplest approach — increasing the number of replicas and sending requests to a random replica. If one replica can handle five hundred requests per second, which is entirely feasible, then three replicas will handle one thousand five hundred.
Sometimes, of course, you can configure ClickHouse for the maximum number of point reads. What is needed for this? First, you need to reduce the granularity of the index. However, it should not be reduced to one but rather based on the assumption that the number of records in the index will be several million or tens of millions on the server. If the table has one hundred million rows, the granularity can be set to 64.
You can reduce the size of the compressed block. There are settings for this. min compress block size, max compress block size. They can be reduced, and the data can be reloaded, which will make point queries faster. However, ClickHouse is still not a key-value database. A large number of small queries is an anti-pattern for load.
Kirill Shvakov: I have a tip in case there are regular counters. It is quite a standard situation when a counter is stored in ClickHouse. I have a user from a certain country, plus some third field, and something needs to be incrementally increased. You take MySQL, create a unique key—in MySQL it’s a duplicate key, while in PostgreSQL it’s a conflict—and add with a plus. This will work much better.
When you have a small amount of data, there’s really no point in using ClickHouse. There are standard databases, and they handle this well.
What to tune in ClickHouse to have more data in the cache?
Let’s imagine a situation—on the servers, there are 256 GB of RAM; in daily routines, ClickHouse takes about 60–80 GB, peaking up to 130. What can be enabled and tuned to have more data in the cache, and accordingly, fewer disk accesses?
As a rule, the operating system’s page cache handles this task well. If you simply open top and look there at cached or free—it also shows how much is cached—you can notice that all available memory is used for cache. And this data will be read not from disk, but from memory. I can say that the cache is used effectively because it caches compressed data.
Nevertheless, if you want to speed up some simple queries even more, there’s an option to enable cache for uncompressed data inside ClickHouse. This is called uncompressed cacheIn the configuration file config.xml, set the uncompressed cache size to the desired value—I recommend no more than half of the available RAM, as the rest will be used for page cache.
Additionally, there are two query-level settings. The first setting is use uncompressed cache which enables its use. It is recommended to enable it for all queries except for heavy ones that may read all data and clear this cache. The second setting is something like the maximum number of rows to use the cache. It automatically limits large queries so that they bypass the cache.
How can we configure storage_configuration for in-memory storage?
In the new ClickHouse documentation, I read a section related to The description includes an example with fast SSD.
I wonder how to configure the same with volume hot memory. And one more question: how does select work with such a data organization? Will it read the entire set or just what is on disk, and are these data compressed in memory? Also, how does the prewhere section function with such a data organization?
This setting affects the storage of data parts, and their format does not change.
Let's take a closer look.
You can configure data storage in RAM. Everything configured for the disk is its path. You create a tmpfs partition that is mounted at some path in the file system. You specify this path as the storage path for the hottest partition, where data parts start to flow in and are written—everything goes well.
However, I do not recommend doing this due to low reliability. Although, if you have at least three replicas in different data centers, it can be acceptable. If something happens, the data will be recovered. Imagine that the server suddenly turned off and then back on. The partition is mounted again, but it's empty. The ClickHouse server, upon starting, sees that these parts are missing, even though, according to ZooKeeper metadata, they should be there. It checks which replicas have them, requests them, and downloads. Thus, the data will be recovered.
In this sense, storing data in RAM is fundamentally no different from storing it on disk, because when data is written to disk, it first goes into the page cache and is physically written asynchronously. This depends on the file system mounting option. However, just for the record, ClickHouse does not perform fsync during an insert.
At the same time, data in RAM is stored in exactly the same format as on disk. The select query retrieves pieces that need to be read in chunks, selects the necessary ranges of data, and reads them. Prewhere operates exactly the same way, regardless of whether the data was in RAM or on disk.
Up to what number of unique values is Low Cardinality effective?
Low Cardinality is cleverly designed. It creates data dictionaries, but they are local. Firstly, the dictionaries are unique for each chunk, and secondly, even within a single chunk, they can vary for each range. When the number of unique values reaches a threshold — I believe it's one million — the dictionary is simply shelved, and a new one is created.
The overall answer is: for each local range — say, for each day — up to about one million unique values, Low Cardinality is efficient. After that, there will simply be a fallback, where many different dictionaries will be used instead of one. It will operate roughly the same way as a regular string type column, perhaps somewhat less efficiently, but there will be no significant degradation in performance.
What are the best practices for full-text search on a table with five billion rows?
There are different possible answers. The first one is to state that ClickHouse is not a system for full-text search. For that, there are specialized systems, for example, and . Nevertheless, I increasingly meet people who say they are transitioning from Elasticsearch to ClickHouse.
Why is this happening? They explain that Elasticsearch begins to struggle with load beyond certain volumes, starting from the indexing aspect. Indexes become too bulky, and if you simply transfer data to ClickHouse, it turns out they are stored several times more efficiently in volume. Moreover, the search queries were often not about finding a phrase throughout the entire data volume considering morphology, but entirely different. For example, finding a byte subsequence in the logs from the last few hours.
In this case, you create an index in ClickHouse, where the first field will be the date with a timestamp. The main data filtering will be based on the date range. Within the chosen date range, you can usually perform a full-text search even using brute force with like. The like operator in ClickHouse is the most efficient like operator you can find. If you find a better one, let me know.
However, like is a full scan. And full scans can be slow not only in terms of CPU but also in terms of disk usage. If you happen to have one terabyte of data daily and you're searching for a specific word within that day, you'll need to scan that terabyte. Likely, it's on regular hard drives, and as a result, they will get so loaded that you won't be able to access that server via SSH.
In this case, I am ready to suggest another little trick. It's experimental—it may work, or it may not. In ClickHouse, there are full-text indexes in the form of trigram Bloom filters. Our colleagues from Arenadata have already tested these indexes, and often they perform exactly as intended.
To use them correctly, it's essential to understand how they work: what a trigram Bloom filter is and how to select its size. I can say that they will help with queries concerning rare phrases or substrings that infrequently appear in the data. In this case, subranges will be selected based on the indexes, and less data will be read.
Recently, ClickHouse introduced even more advanced features for full-text search. First, it allows for searching multiple substrings in a single pass, including case-sensitive, case-insensitive options, supporting UTF-8 or just ASCII. Choose the most efficient one that you need.
There is also the ability to search for multiple regular expressions in one pass. You don't need to write X like one substring or X like another substring. Just write it all at once, and everything will execute as efficiently as possible.
Thirdly, there is now approximate regex search and approximate substring search. If someone types a word with a typo, it will be searched for maximum matching.
How can we better organize access to ClickHouse for a large number of users?
Explain how to best organize access for a large number of consumers and analysts. How to form a queue, prioritize requests for max concurrent queries, and what tools to use?
If the cluster is large enough, a good solution is to set up two additional servers that will serve as entry points for analysts. This means not allowing analysts direct access to specific shards of the cluster, but rather creating two empty servers without any data, and configuring access rights on those. In this case, user settings for distributed queries are transferred to the remote servers. So, you configure everything on these two servers, and the settings take effect across the entire cluster.
These servers are essentially data-free, but the amount of RAM on them is crucial for processing queries. The disk can also be utilized for temporary data if external aggregation or external sorting is enabled.
It is important to look at the settings associated with all possible limits. For example, if I access the 'Yandex.Metrics' cluster as an analyst and pose a query select count from hits, I will immediately receive an exception stating that I cannot execute the query. The maximum number of rows I am permitted to scan is one hundred billion, while the entire cluster has fifty trillion in one table. This is the first limitation.
Let's say I remove the limit on the number of rows and execute the query again. Then I will see the next exception— the setting force index by dateis enabled. I cannot execute the query if I do not specify a date range. One should not rely on analysts to specify it manually. A typical case would be to write a date range where event date between last week. Then, just placing a bracket incorrectly, instead of 'and', it became 'or' — or URL match. If there are no limits, it will start scanning the URL column and waste a ton of resources.
Moreover, ClickHouse has two priority settings. Unfortunately, they are very primitive. One is simply called priorityIf priority ≠ 0, and requests are being processed with some priority, but a request with a lower priority value (which means a higher priority) is being processed, then the request with the higher priority value (indicating a lower priority) is simply suspended and will not execute during this time.
This is a very rough setting, and it is not suitable for scenarios where the cluster has a constant load. However, if you have short, bursty important requests, and the cluster is mostly idle, this configuration will work.
The next priority setting is called OS thread priority. It simply sets the nice value for all threads managing requests in the Linux scheduler. It works somewhat, but still works. If the minimum nice value is set – which is the largest numerically, meaning the lowest priority – and for high-priority requests, a value of -19 is set, then the CPU will consume low-priority requests about four times less than high-priority ones.
You also need to configure the maximum execution time for a request – say, five minutes. The minimum execution speed is a critical setting. This setting has been around for a while and it is necessary not just to claim that ClickHouse does not lag, but to enforce it.
Imagine you are configuring: if any request processes less than one million rows per second – this is unacceptable. It tarnishes our good name, our excellent database. Let's just ban this. There are actually two settings. One is called min execution speed – in rows per second, and the second is called timeout before checking min execution speed – by default, fifteen seconds. This means that fifteen seconds is allowed, but then, if it is slow, simply throw an exception – terminate the request.
You also need to set up quotas. ClickHouse has a built-in quota feature that tracks resource consumption. However, unfortunately, it does not count physical resources like CPU, disks, but logical ones – the number of processed requests, rows, and bytes read. For example, you can set a maximum of one hundred requests over five minutes and a thousand requests per hour.
Why is this important? Because part of the analytics queries will be executed manually from the ClickHouse client. And everything will be fine. But if your company has advanced analysts, they will write a script, and there could be an error in the script. This error could lead to the query running in an infinite loop. That is what needs to be protected against.
Can we send the results of one query to ten clients?
We have several users who like to come with very large queries at the same time. The query is large, it generally executes quickly, but due to the number of such queries running simultaneously, it becomes very painful. Is it possible to execute the same query that came ten times in a row just once, and then return the result to ten clients?
The problem is that we don't have cached results or intermediate data caches. There is the operating system's page cache, which prevents data from being read from the disk again, but unfortunately, the data still needs to be decompressed, deserialized, and processed again.
We would like to avoid this somehow, either by caching intermediate data or by organizing similar queries into a queue and adding a results cache. Currently, we have a pull request in development that adds query caching, but only for subqueries in the IN and JOIN sections — meaning the solution is incomplete.
Nevertheless, we also face a similar situation. A particularly canonical example is pagination queries. There is a report that has several pages, and it issues a query with limit 10. Then the same again, but with limit 10,10. Then the next page. And the question arises, why do we calculate all of this each time? But for now, there is no solution, and there's no way to avoid this.
There is an alternative solution that can be set up as a sidecar next to ClickHouse — .
Kirill Shvakov: ClickHouse Proxy has a built-in rate limiter and integrated results cache. It has many configurations because it addresses a similar task. The Proxy allows you to limit queries by organizing them into a queue and configuring how long the query cache lasts. If the queries were indeed identical, the Proxy will serve them multiple times while only going to ClickHouse once.
Nginx also has a cache in the free version, and it works too. Nginx even has settings to throttle other requests if they come in simultaneously, waiting for one to complete. However, in ClickHouse Proxy, this setting is implemented much better. It was specifically built for ClickHouse and its queries, making it more suitable. And it's easy to install.
How do we handle asynchronous operations and materialized views?
There is a problem where operations with the replacing engine are asynchronous—first, data is written, and then it is compacted. If there is a materialized view below the table with some aggregates, duplicates will be written to it. If there is no complex logic involved, the data will be duplicated. What can be done about this?
An obvious solution is to implement a trigger on a specific class of materialized views during the asynchronous compaction operation. Are there any 'silver bullets' or plans to implement such functionalities?
It is important to understand how deduplication works. What I am about to discuss is not directly related to the issue at hand, but it's worth mentioning it just in case.
When inserting into a replicated table, there is deduplication of entire inserted blocks. If you re-insert the same block containing the same number of the same rows in the same order, the data will be deduplicated. You will receive 'Ok' in response to the insert, but effectively only one batch of data will be recorded, and it will not be duplicated.
This is necessary for clarity. If you receive 'Ok' during insertion, your data has been inserted. If you receive an error from ClickHouse, then they were not inserted, and the insertion needs to be retried. But if the connection was lost during the insertion, you cannot know whether the data was inserted or not. The only option is to repeat the insertion again. If the data was indeed inserted, and you try to insert it again, there will be deduplication of blocks. This is needed to avoid duplicates.
It is also important how it works for materialized views. If the data was deduplicated when inserted into the main table, it will not go into the materialized view either.
Now, regarding the issue. You have a more complex situation because you are recording duplicates of individual lines. That is, not an entire batch is duplicated, but specifically certain lines, and they collapse in the background. Indeed, the data will collapse in the main table, and the non-collapsed data will go into the materialized view, and during merges, nothing will happen to the materialized views. Because a materialized view is nothing more than a trigger on an insert. Nothing additional happens to it during other operations.
And I can't really cheer you up here. It’s necessary to look for a specific solution for this case. For example, is it possible to do a replacement in the materialized view too, and maybe the deduplication method will work the same way. But unfortunately, it's not always the case. If it is aggregating, then it won't work.
Kirill Shvakov: We too had our share of workarounds in the past. There was a problem that there were ad impressions, and there are some data we can show in real time — these are just displays. They rarely get duplicated, but if that happens, we will still collapse them later. And there were things that couldn't be duplicated — clicks and all that. But we wanted to show them almost immediately.
How were the materialized views created? There were views where data is written directly — there's a record in raw data, and it's written into views. At some point, the data isn't very correct, they duplicate, and so on. And there’s a second part of the table, where they look exactly the same as the materialized views, meaning they are absolutely identical in structure. Once in a while, we recalculate the data, counting them without duplicates, and write to those tables.
We went through the API — working directly with ClickHouse won't work. The API checks: when I have a date of the last addition to the table, where the data is guaranteed to be accurate, calculated, it makes a request to one table and another. From one, it selects up to a certain amount of time, and from the other, it gathers what hasn’t been calculated yet. And it works, but not with the means of just one ClickHouse.
If you have an API for analytics or users, then this is basically an option. You always count, you always recalculate. This can be done once a day or at some other time. You choose the range that is not critical for you.
ClickHouse has a lot of logs. How can I see everything that happens with the server in real-time?
ClickHouse has a very large number of different logs, and this number is increasing. In new versions, some of them are even enabled by default, while in older versions, they need to be turned on during updates. Nevertheless, there are more and more of them. I would like to see, in the end, what is currently happening with my server, perhaps on some consolidated dashboard.
Do you have a ClickHouse team, or do your friends have teams that support any functionality of ready-made dashboards that display these logs as a finished product? In the end, it's great to look at logs in ClickHouse. But it would be really cool if it were already prepared as a dashboard. I would enjoy that.
Dashboards do exist, although they are not standardized. In our company, about 60 teams use ClickHouse, and the strangest thing is that many of them have dashboards that they created themselves, and they are slightly different. Some teams use an internal installation of Yandex.Cloud. There are some ready-made reports there, although not all necessary ones. Others have their own.
My colleagues from Metrics have their own dashboard in Grafana, and I have my own based on their cluster. I look at things like cache hits for the matches cache. It's even more complicated because we use different tools. I created my dashboard with a very old tool called Graphite-web. It is quite ugly. And I still use it, although Grafana would probably be more convenient and nicer.
The basic thing in dashboards is the same. These are system metrics for the cluster: CPU, memory, disk, network. Others include the number of concurrent requests, the number of concurrent merges, the number of requests per second, the maximum number of parts for MergeTree table partitions, replication lag, the size of the replication queue, the number of rows inserted per second, and the number of blocks inserted per second. This is all that is obtained not from logs, but from metrics.
Vladimir Kolobaev: Alexey, I would like to make a slight correction. There is Grafana. Grafana has a datasource, which is ClickHouse. This means I can make queries directly to ClickHouse from Grafana. In ClickHouse, there is a table with logs, which is the same for everyone. I want to be able to access this logs table in Grafana and see the queries that my server sends. It would be great to have such a dashboard.
I put it together myself. But I have a question — if everything is standardized and Grafana is used by everyone, why is there no official dashboard for it at Yandex?
Kirill Shvakov: Actually, the datasource that connects to ClickHouse is currently supported by Altinity. I just want to give direction on where to dig and whom to push. You can ask them, because Yandex is indeed the one making ClickHouse, not the surrounding ecosystem. Altinity is the main company currently promoting ClickHouse. They will not abandon it but will continue to support it. Because in principle, to upload a dashboard to the Grafana site, you only need to register and upload it — there aren’t really any problems.
Alexey Milovidov: Over the past year, many capabilities for profiling queries have been added to ClickHouse. There are metrics for each query regarding resource usage. And just recently, an even lower-level query profiler has been added to see where each millisecond of the query is spent. However, to take advantage of this functionality, I have to open the console client and type the query, which I constantly forget. I saved it somewhere and I keep forgetting where exactly.
I wish there were a tool that simply stated — here are your heavy queries, grouped by query classes. I could click on one, and it would tell me why it is heavy. Currently, there is no such solution. It's quite strange that when people ask me, 'Are there any ready-made dashboards for Grafana?', I reply, 'Go to the Grafana website, check the community section for
How can I influence merges so that the server does not fall into OOM?
I have a table that contains only one partition, which is ReplacingMergeTree. I have been writing data to it for four years. I needed to make an alter and delete some data.
I did this, and during the processing of this request, all the memory was consumed across all servers in the cluster, and all cluster servers went into OOM together. Then they all started to recover, began to perform the merge of the same operation, this block of data, and fell into OOM again. Then they recovered again and fell once more. This cycle did not stop.
It then turned out that this was actually a bug that the team fixed. That's great, thank you very much. But the residue remains. Now, when I think about needing to do some merge in the table, I question why I can't somehow influence these merges? For example, limit them in terms of the required RAM or even the number of them that will specifically process this table.
I have a table called ‘Metrics’, please process it for me in two streams. Do not spawn ten or five merges in parallel; do it in two. I believe that two will suffice in terms of memory, while processing ten may not. Why does the fear remain? Because the table is growing, and someday I will encounter a situation where, not due to a bug, but because the data will change in such large quantities, I simply won’t have enough memory on the server. Then the server will go into OOM during the merge. The mutation can be canceled, but the merges cannot.
You know, during merges the server will not crash due to OOM, because the memory used during a merge is only for a small range of data. So everything will be fine regardless of the data volume.
Vladimir Kolobaev: Okay. There's this point that after fixing the bug, I downloaded the new version, and on another smaller table, where there are many partitions, I performed a similar operation. During the merge, the server consumed about 100 GB of RAM. I had 150 GB occupied, 100 GB was used, leaving me with a window of 50 GB, so I didn't hit OOM.
What currently protects me from hitting OOM, if it truly consumes 100 GB of RAM? What should I do if suddenly the RAM runs out during merges?
Alexey Milovidov: There is a problem that the RAM consumption during merges is not limited. The second problem is that if a merge has been scheduled, it must be executed because it's recorded in the replication log. The replication log contains the actions necessary to bring the replica to a consistent state. If manual actions are not taken to revert this replication log, the merge will have to be executed one way or another.
Of course, it would be helpful to have a RAM limit that protects against OOM 'just in case.' It won't help the merge complete; it will start over, reach a certain threshold, throw an exception, and then start again – nothing good will come of it. But in principle, introducing such a limit would be useful.
How will the development of the Golang driver for ClickHouse take place?
The Golang driver written by Kirill Shvakov is now officially supported by the ClickHouse team. , it is now large and real.
A small remark. There is a wonderful and universally loved storage for flexible, infinite order — Vertica. They also have an official Python driver supported by the Vertica developers. There have been times when the versions of the storage and the driver diverged significantly, causing the driver to stop working at some point. And the second point: support for this official driver seems to operate on a 'nipple' system — you write them an issue, and it hangs forever.
I have two questions. Currently, Kirill's Golang driver is almost the default way to communicate from Golang to ClickHouse. Unless someone prefers to communicate through the HTTP interface because they like it that way. How will the development of this driver be managed? Will it synchronize with any breaking changes in the storage itself? And what is the process for issue resolution?
Kirill Shvakov: First of all, how everything is organized bureaucratically. This point has not been discussed, so I have nothing to answer.
To answer the question about issues, a brief history of the driver is needed. I worked at a company that dealt with a lot of data. It was an advertising platform with an enormous number of events that needed to be stored somewhere. At some point, ClickHouse emerged. We started pouring data into it, and everything was fine initially, but then ClickHouse crashed. At that time, we decided we didn't need it.
A year later, we returned to the idea of using ClickHouse, and we needed a way to write data to it. The premise was that the hardware was very weak, with limited resources. But we always worked this way, so we looked towards the native protocol.
Since we were working in Go, it was clear that we needed a Go driver. I worked on it almost full-time — it was my job. For a while, we got it to a workable state, and basically, no one expected anyone other than us to use it. Then CloudFlare came along with exactly the same problem, and for some time, we collaborated closely with them because they had the same challenges. Moreover, we worked both on ClickHouse itself and on the driver.
At some point, I just stopped dealing with them because my activity regarding ClickHouse and my work has changed a bit. Therefore, issues are not being closed. Periodically, people commit to the repository when they need something. Then I look at the pull requests, and sometimes I even make corrections myself, but that happens rarely.
I want to get back to the driver. A few years ago, when all of this started, ClickHouse was different and had different capabilities. Now there is an understanding of how to redesign the driver to make it better. If that happens, version 2 will definitely be incompatible due to accumulated hacks.
I don't know how to organize this. I don't have much time myself. If some people will be working on the driver, I can help them and explain what to do. But active participation from Yandex in the project's development has not been discussed yet.
Alexey Milovidov: In fact, there is still no bureaucracy regarding these drivers. The only thing is that they have been moved to an official organization, meaning this driver is recognized as the official default solution for Go. There are some other drivers, but they are separate.
We do not have any internal development for these drivers. The question is whether we can hire a dedicated person, not specifically for this driver, but for the development of all community drivers, or if we can find someone externally.
The external dictionary does not load after a restart with the lazy_load setting enabled. What should I do?
We have the lazy_load setting enabled, and after restarting the server, the dictionary does not load automatically. It only loads when a user accesses this dictionary. And at the first access, it gives an error. Is there a way to automatically load dictionaries using ClickHouse, or do we always need to monitor their readiness ourselves so that users do not encounter errors?
Perhaps we have an old version of ClickHouse, which is why the dictionary did not load automatically. Is that possible?
First of all, dictionaries can be forcefully loaded using the query system reload dictionaries. Secondly, regarding the error — if the dictionary is already loaded, the queries will work with the data that was loaded. If the dictionary has not been loaded yet, it will load at the time of the request.
For heavy dictionaries, this is not very convenient. For example, if you need to pull a million rows from MySQL. Some people do a simple select, but that select will wait for those million rows. There are two solutions. The first is to turn off lazy_load. The second is that when the server starts up, before putting it under load, do system reload dictionary or simply execute a query that uses the dictionary. Then the dictionary will load. You need to manually control the availability of dictionaries with lazy_load enabled, as ClickHouse does not automatically fetch them.
To the last question, the answer is that either the version is old, or it needs debugging.
What to do if system reload dictionaries does not load any of the many dictionaries, if even one of them fails with an error?
There’s also a question regarding system reload dictionaries. We have two dictionaries—one does not load, the other does. In this case, system reload dictionaries does not load either dictionary, and we have to specifically load one by its name using system reload dictionary. Is this also related to the ClickHouse version?
I want to give you good news. This behavior has changed. So, if you update ClickHouse, it will also change. If you are not satisfied with the current behavior system reload dictionaries, update, and let’s hope it improves.
Is there a way to configure credentials in the ClickHouse config without exposing them in case of errors?
The next question is about errors related to the dictionary, namely, credentials. We specified the connection credentials in the ClickHouse config for the dictionary, and in case of an error, we receive these credentials and the password in the response.
We resolved this issue by extracting the credentials into the ODBC driver config. Is there a way to configure the credentials in the ClickHouse config without exposing them during errors?
Here, the solution is indeed to specify these credentials in odbc.ini, and in ClickHouse itself, only specify the ODBC Data Source Name. This will not be the case for other dictionary sources—neither for the MySQL dictionary nor for others should you see the password in the error message. I’ll check for ODBC as well—if such a thing exists, it should just be removed.
Bonus: backgrounds for Zoom from gatherings
By clicking on the image, the most persistent readers will unlock bonus backgrounds from gatherings. We’re extinguishing fires together with the Avito technology mascots, having discussions with colleagues from the sysadmin room or an old-school computer club, and holding daily meetings under the bridge against a backdrop of graffiti.
Source: habr.com
