Strength in Numbers. Introduction to ClickHouse® Cluster Performance

Recorded: April 29, 2020
Presenters: Robert Hodges and Alexander Zaitsev

In this webinar, Altinity CEO Robert Hodges and CTO and co-founder Alexander Zaitsev introduce the performance characteristics of ClickHouse clusters, aimed at beginning-to-intermediate users who want to get more out of distributed deployments. They explain the two axes of horizontal scaling, sharding to add read and write capacity over a data set and replicas to add concurrent read capacity, and contrast them with the vertical scaling of a single node. Replication is multi-master and asynchronous, coordinated through ZooKeeper, which tracks the parts that need to move between replicas.

Using a distributed copy of the airline on-time data set, Robert walks through the practical setup: defining a sharded, replicated fact table with ReplicatedMergeTree and ON CLUSTER, using macros so one DDL statement works cluster-wide, layering a distributed table on top, and choosing between inserting through the distributed table or directly to the underlying shards. Direct insertion to shards generally gives the best performance and avoids the extra buffering and data-loss risk of asynchronous distributed inserts. He covers sharding-key choices, including random keys, keyed sharding by tenant, and a bi-level scheme that combines both, and notes that ClickHouse does not automatically rebalance data.

On the read side, distributed queries push as much work as possible down to the shards, returning partial aggregate states to the initiator node for merging, which yields close to linear speedups when queries are heavy enough to outweigh network overhead. Alexander explains distributed_product_mode and how joining two distributed tables can run in local or global form, with global being the safe default. Robert closes with capacity-planning guidance drawn from Mik Kocikowski’s CloudFlare talk, a list of things that commonly go wrong in clusters, and a reminder that good cluster design should produce dramatically faster results, followed by an extended Q&A with Alexander.

Here are the slides:

Key Moments (Timestamps)

Key moments generated with AI assistance.

• 0:13 – Introduction and housekeeping
• 1:47 – Presenters and goals of the talk
• 4:50 – A brief introduction to ClickHouse
• 6:28 – Vertical and horizontal scaling
• 10:18 – Sharding and replication patterns
• 15:38 – Setting up the airline dataset
• 20:26 – Inserting data: distributed vs local
• 27:04 – Choosing a sharding key
• 34:12 – How distributed queries work
• 38:16 – Distributed product modes and joins
• 44:38 – Capacity planning and what goes wrong
• 49:34 – Key takeaways
• 52:56 – Q&A

Webinar Transcript

0:13 — Introduction and Housekeeping

Robert Hodges: Hi everybody, this is Robert Hodges, and I’d like to welcome you to our webinar, Strength in Numbers: an introduction to ClickHouse cluster performance. Before we dive in, I’d like to tell you a few things about how we’re organizing this today, which may save you time and increase your enjoyment of the material. First, this webinar is being recorded, and we’ll send both the recording and a link to the slides afterwards, so you don’t have to frantically take notes. Second, we have a question-and-answer box, and we invite you to pop questions into it as things occur to you; if the topics are relevant to what we’re talking about at the time, we’ll probably take them in line, otherwise we’ll have time at the end to answer questions at some length, hopefully at least 10 or 15 minutes, because this is a complex topic and many of the issues you have won’t necessarily be covered on the slides. Finally, when we start the Q&A, I’ll be running a poll, as we do every time, to get feedback on how we did, where you’re at in your ClickHouse usage, and what other topics you’d like to see in the future. With that, I’m going to go ahead and jump in.

1:47 — Presenters and Goals of the Talk

Robert Hodges: A little about myself: I’m the CEO of Altinity, and I’m also a database geek; I’ve been working on databases of one kind or another since 1983, and ClickHouse is about number 20. I’m also delighted to say I have with me today our CTO, Alexander Zaitsev. Alexander, do you want to check in and introduce yourself?

Alexander Zaitsev: Hi everybody, it’s nice to see a good crowd here. I am CTO and founder of Altinity, and we’re making ClickHouse better for you, so I’ll be helping Robert with some of the material.

Robert Hodges: Thank you, Alexander. Alexander will make this webinar better because there’s some material we’ll touch on where he has unique experience. One more thing before we dive in, just a little about Altinity: Alexander was a founder, and our goal is to make people successful with ClickHouse. We’re the leading provider of software and services for ClickHouse and a major committer as well as a community sponsor in the US and Western Europe. I’m glad to see that some of the people here are already our customers; welcome, and we hope everybody else will follow suit.

In this talk we’re going to discuss the performance of ClickHouse clusters. We’ll dig into how distributed clusters are set up, talk about using shards to scale writes and reads, and using replicas to scale read concurrency, so you understand the architecture and the axes you can use to scale, and then we’ll spend time going through tricks, performance bottlenecks, and other things you need to know to use distributed clusters effectively. Two things we’re not going to do: we won’t talk about the performance of single nodes, which we covered in a popular earlier webinar that dives deeply into the behavior of the MergeTree table and how to optimize a single node; we’ll take that as a given here. And we’re not going to teach advanced ClickHouse performance management; this talk is aimed at beginner-to-intermediate users of ClickHouse clusters, to help you be more productive.

4:50 — A Brief Introduction to ClickHouse

Robert Hodges: For those of you who are new to ClickHouse, I always like to give a brief introduction. I often describe it as follows: imagine MySQL, a very popular open source database, and Vertica, a very popular proprietary data warehouse, got married and had a baby; that baby would be ClickHouse. From the MySQL side of the house, you’re open source, you understand SQL, you run practically anywhere, and you’re pretty easy to manage, and at least in the design sense those properties have been inherited by ClickHouse. From the Vertica side, you have column organization, vectorized execution locally as well as parallel execution across nodes, the ability to scale to many petabytes, and you’re extremely fast on OLAP queries. Take those two things together and that basically gets you ClickHouse. ClickHouse was originally developed at Yandex; the design started in 2009, and it was open sourced in 2016. From here on out, we’re going to take it as a given that you have some experience with ClickHouse, and what I’ll do now is dive into the cluster model and how things are organized.

6:28 — Vertical and Horizontal Scaling

Robert Hodges: Before I get too deeply into that, it’s good to talk about ClickHouse nodes and understand the scaling model we begin with when we run a single node. There are basically three ways you can increase the capability, and we call this vertical scaling, because instead of adding more nodes we’re making the nodes we have more capable: you can add more memory, add more CPUs, or upgrade your storage. Storage is shown here as a single axis, but it’s more accurately multiple axes, because you can scale the amount of storage and the number of IOPS. So those are three general ways to change the capacity of a ClickHouse node, going from low capacity up to a great deal.

As we introduce clusters, instead of scaling vertically we’re going to scale horizontally, and there are two basic ways ClickHouse does this that are baked into the model. The first is sharding, which means you take a data set and divide it into separate pieces. For example, in a multi-tenant system you could split it out by tenant and have a node for each one, or if you were analyzing marketing data you could arbitrarily divide it into thirds and put a third on each node. What sharding does is add IOPS, in the same way that when you apply RAID you bind together a bunch of spindles and get the aggregate performance of all those devices; by adding shards you can write to more data because you have more places to write, and read more data because you have more nodes to help with reads. The other way to extend horizontally is to add replicas. Having a lot of replicas doesn’t really help individual query performance, but it’s exceedingly helpful if you have a lot of concurrent users, since it gives them more places to go to select data so they’re not treading on each other. These are the two forms of horizontal scaling ClickHouse supports very well.

There’s a question here: what’s usually one of the triggers to move from a single-node approach to a cluster approach? Great question. In a nutshell, if you can go single node, generally you’ll move to a cluster for two reasons: one is that you think you’re going to exceed the capacity, either now or in the near future, say a 12-month horizon; that’s a great reason. Another reason to horizontally scale, particularly around replicas, is for high availability; we’re not really going to talk about HA, but that’s another reason people run replicas.

10:18 — Sharding and Replication Patterns

Robert Hodges: One thing that’s really important to understand about ClickHouse is that when we talk about sharding and replication, it’s always at the table level. We have a bunch of nodes that can support replication and sharding, but the decisions about which data goes where are actually made at the table level, and there are three general patterns. One is to have everything be a shard; a good example is the system tables, like the query_log, which is exceedingly useful for tracking your queries. You can think of that as having a shard present on each node with no replication, and if you want to look at the queries across the entire system, you can put a distributed table on top so you can see all the query_log contents across the system. Another approach is all replicated, where there’s a copy of the table on every node; a good example is dimensions that are used universally, sometimes relatively small tables that you put on every node so you can join locally on them, which we’ll discuss later. Finally, there’s sharded and replicated, where you split the data into pieces and then have some number of replicas for each piece; this is a common pattern for very large fact tables covering sales, events, or marketing campaigns.

One thing that’s important to understand is that there’s nothing magic about these patterns; you can define them any way you want. The way they’re actually implemented is through a file we call, by convention, remote_servers.xml, which has a remote_servers tag as part of the ClickHouse server configuration, and it just lists hosts with different ways to arrange them. In this case we have a sharded and replicated layout, which is also sometimes referred to as a cluster: we have a shard that consists of two hosts, another shard of another two hosts, and so on. The clustering pattern is basically what you say it is, and the types I described are just conventions; overall it’s simpler if you use relatively standard patterns, but you can use any layout you want.

One final thing on replication: ClickHouse replication is multi-master and asynchronous, which means that if you load data into a replica, that results in a record made in ZooKeeper, which keeps the list of parts that need to be replicated for that set of replicas, and then the other replicas will, in a short time, notice it’s there, go talk to the original server, grab the part, and pull it across. So when you use replication, you can insert into any one of your nodes and the data will eventually be replicated to the other nodes. There are ways to alter this behavior; for example, you can force an insert to become fully synchronous, or set a quorum so that you know on the application side that the data successfully reached more than one replica. We did a webinar back in the summer that covers the implementation details of replication in quite a bit of detail.

15:38 — Setting Up the Airline Dataset

Robert Hodges: That’s the theory; now I’d like to move into the practice, which is where it gets interesting. To practice, it’s helpful to have a sample, so I set up one of our favorite data sets, the airline on-time data set; my copy has about 173 million rows. Normally we run this on a single node because it’s really small, a great example of something you don’t need a cluster for, but I’ve set it up as a distributed data set for testing. We have the shards of data, shown in red; there are two shards, each replicated two ways, in a table called on_time_shard, which holds our chunks of the airline fact table. We have a dimension table called airports that’s replicated fully across all nodes, and then to query these we have a distributed table called on_time, which lets us both load and query data in the underlying on_time_shard tables.

For the fact table, one command you want to get really familiar with when running clusters is CREATE TABLE ON CLUSTER, or CREATE DATABASE ON CLUSTER. This creates a table called on_time_shard, and the ON CLUSTER means it gets propagated across whatever your cluster is defined to be, in this case sharded and replicated. We’re using ReplicatedMergeTree, which sets up an engine and a path that tells ClickHouse, through ZooKeeper, where the information about the state of this particular table is and what needs to be replicated. The other parts, PARTITION BY flight date and ORDER BY, are normal things you see in any MergeTree table. You may have noticed I used macros, the little things with squiggly brackets; to make clusters work, it’s really important to set up macros, which let you use the same definition across the entire cluster, with ClickHouse filling in the values.

Since we’ve got the fact table defined with a nice sharding pattern, we set up a distributed table next, which is a much simpler piece of DDL because we can copy the schema from the on_time_shard table. We use the Distributed table engine, where we tell it the cluster the table belongs to, the database where the data is located (airline shards), and the name of the underlying shard table, and then the final argument is a sharding key. When we insert into the distributed table, this enables it to decide which shard to put data in; in this case we’re using rand(), which just randomly assigns each incoming row to a shard, a reasonable way to distribute things for small data sets and small clusters. Finally, the replicated dimension table uses the same replicated engine; it’s relatively small, so we PARTITION BY tuple() so ClickHouse doesn’t bother dividing it into parts, and we use a cluster called all_replicated, which simply lists all servers in the cluster.

20:26 — Inserting Data: Distributed vs Local

Robert Hodges: With our sample set up, let’s insert some data and see how fast it is. One really important decision in loading data into a cluster, and getting the best performance, is whether to insert locally to the shards (these are just ordinary tables, so you can put data straight into them) or use distributed approaches. One popular distributed approach is to go through the distributed table, using the sharding function to split the data. Another, which is very popular but we won’t touch on deeply, is to use chproxy, which load-balances the transactions across nodes. By default, inserting into a distributed table is an asynchronous operation, which is good because you can blast the data in and run off to do something else, but there are cases where you want to make sure it really arrived, so insert_distributed_sync and insert_quorum are options that let you ensure the distributed table isn’t batching things behind your back and that replication actually took place. For this talk we’ll use the asynchronous defaults, but those settings are available if you care.

Let’s look at the simple case of inserting through the distributed table versus the data pipeline. If we point our data pipeline at the distributed table, we can pick any of these distributed tables because I made them across the entire cluster, and we just insert. By default, the on_time distributed table engine grabs these inserts, breaks them up using the sharding function into separate shards, sticks them in a temporary location, and then ensures they’re pushed, or more correctly pulled, across to the right place where they need to land. If you’re watching, you can see these files piling up and eventually reaching the two shards. The other way is to have your data pipeline understand your sharding pattern and point straight at the underlying tables, because there’s nothing magic about shards in ClickHouse, they’re just tables; what brings a set of shards together is a distributed table on top that can see them all. As a result, many people build their data pipelines to go straight to the underlying tables; you just pick one replica in each case, blast the data in, and it gets replicated out properly to the remaining replicas. There are trade-offs: going directly means your pipeline needs to be shard-aware, but inserting into the distributed table is slower because the data is rewritten, and it can be more error-prone.

If we test this, we see results like the following. I tried a single shard using MergeTree, so it wasn’t replicated at all, then two shards replicated two ways going to the distributed table, and then direct loading to the shards without going through a distributed table. If your systems have lots of capacity, it’s about equal loading time for all of these. Where it gets interesting is as your volume goes up, which is probably why you’re clustering. As I increased the load in the single-node case, I loaded with 20 threads to push performance and load faster; when I did that to the distributed table, it crashed, because going to the distributed table I’m having to buffer this and it consumes more resources, and in this case I didn’t have enough. That is not a big problem overall, lots of people run things through distributed tables, but it illustrates that it requires more resources. With direct loading I was able to get the best performance, going to the two replicated shards directly, and with more shards you’d see a much more pronounced improvement. So when using distributed tables, be aware that they’re somewhat resource-inefficient because they’re double-copying the data, and because they run asynchronously there’s a chance you can lose data if a shard crashes, since you’re not guaranteed all the data reaches its location. This is also something you definitely want to monitor carefully, because the behavior underneath is not super visible.

27:04 — Choosing a Sharding Key

Robert Hodges: One really big design question people go through, in fact I was on a call about this earlier this morning, is selecting the sharding key. The experience of many large sites is that they took a decision and then regretted it and had to change it later, so it’s worth thinking about upfront. There are two general approaches. One is to randomly shard: pick some value X in a column that varies pretty randomly, like an arrival date or a session key, some high-cardinality value, and run a cityHash on it to split it across, in this example, three shards. The other way, in a multi-tenant system, is to take the modulo of a tenant ID and shard by tenant. There are trade-offs. With the random key, the big downside is that you typically have to query all the shards, because ClickHouse has no way of knowing where the rows went in a way it can trace back to the query. With sharding by tenant, that’s not a problem because you can begin to skip shards, either explicitly in the application or, increasingly, with ClickHouse’s own capabilities, but you’re going to get unbalanced nodes, which is very hard to control because you never know how much data you’ll have for a particular tenant; some tenants do well, some don’t, or may go away, so you’ll have rebalancing issues. One key thing is that ClickHouse does not automatically rebalance data.

So what some people have done for very large installations is a bi-level sharding approach, where you take the top level, the tenant ID, and shard off it to get a group, and then within the groups you use a random approach that spreads the data evenly. So for some tenant X, they end up in one of the groups, and at the very least you’d only have to check, say, three servers instead of seven in this example. This gives you the best of both worlds because you can add nodes more easily within the groups, but also have some ability to control the space you traverse for queries. One thing important for query performance: the more nodes your query touches, the more chances one of them is going to be slow, and in distributed queries the slowest node typically determines your performance, so keeping that number reasonable means better overall performance.

To repeat, there’s no magic in the sharding schemes; if you pick something like bi-level sharding, you just add more macros and more stuff in the path of the ReplicatedMergeTree. Any sharding and replication scheme you decide to adopt, you implement by appropriate changes to the macros file, the remote_servers, and the ReplicatedMergeTree settings, so it’s pretty flexible. By the way, for the number used to compute the modulo in bi-level sharding, that’s a somewhat synthetic example; you would decide a number in advance, or it could be a lookup at the front where you map tenant IDs to particular groups, so there’s a variety of ways, again no magic, just something your applications need to understand.

On adding nodes and rebalancing data, ClickHouse doesn’t magically move data around, so as you build the cluster over time you’ll need to add capacity to maintain performance. The random sharding schemes allow easier addition of shards, and in particular if you have time-series data that you keep for, say, seven days using a TTL, you can bring more nodes into the cluster, assign them to cover particular shards, and as your time-series data arrives, it slowly fills up that node, and after seven days all the shards are balanced; that’s the easiest way to do it, and replication propagates the dimension tables. If that doesn’t work, because you don’t have conveniently rebalancing data or you’re using a key partitioning scheme, you’ll end up having to move data yourself, and there are no magic tools for that yet, though we’re working on it. One of the big things about ClickHouse is that data is immutable; once a part is written it typically doesn’t get rewritten, which makes it fairly straightforward to move data around.

34:12 — How Distributed Queries Work

Robert Hodges: Let’s talk about distributed queries, where the behavior gets really interesting. To get a query to run across multiple shards, we come through a distributed table, so the application writes to the on_time table, which we call the initiator node. It processes the query initially and tries to push down as much of it as it can; the innermost sub-select, the left side of the query, gets pushed down, and on the actual shards it computes the partial aggregates, known as aggregate state functions, so it can do counts and things like that in a partially aggregated form. Those are then returned to the initiator node, which merges them. One thing you can see is that the more work that has to be done on the single initiator node, the slower things get and the more resources you need, so to make distributed queries really work for you, push down the WHERE and your heaviest grouping work into the innermost query and get it done locally, then just the results come back to the initiator.

Here’s a simple example with no join: I ran a fairly thick query that forced us to scan a bunch of columns, and if you do this right you get linear performance in the number of nodes. A single shard took about 10 seconds cold; the second run was hot, with pages from storage stored in the buffer cache, and running one to four shards you see basically linear performance. One thing to notice is that if you have very fast queries, at some point the network overhead dominates your parallelization, so below a certain level you won’t see improvements because you still have to transfer the results around and put them together, but when you’re doing this well the results are pretty phenomenal. By default, when you come into the distributed table, ClickHouse pushes down the joins; if you have a right and left join, that gets pushed out into the lower table. You can see this in the query logs: the initial query lands on the distributed table node, and on the replicas that get picked you’ll see the pushdown query, where ClickHouse changed on_time, the distributed table, to on_time_shard and ran it there. If you don’t want that, a simple way is to make the left-side query a sub-query, so only that gets pushed down and the join happens later; in this case that was beneficial because we were joining on the airports, which was faster to do once we had aggregated the data.

38:16 — Distributed Product Modes and Joins

Robert Hodges: At this point things have been pretty straightforward, but there are a number of settings that can further affect joins, and for those I’m going to invite Alexander to talk about distributed product modes, how they affect joins, and the effect on IN operators.

Alexander Zaitsev: Thank you, Robert. I’ll start by answering a question from our questionnaire: if a huge dimension table needs to be joined, you can put it on a single shard, but then that table is only available on that shard, so to use it in a join you need to replicate it; that’s pretty straightforward. The real interesting stuff starts if you want to join two distributed tables. For example, you have one distributed table and you want to join another distributed table, and it’s actually not evident how you want to do it. Robert, can you show the next slide? Consider that you have two tables, t1 and t2, and you want to join them, or it’s pretty much the same if you want to do an IN condition. In this case you have three different ways the query can be executed. First, you can run the local table on every node, both t1 local and t2 local, and that will produce a correct result if both tables are sharded by the same sharding key, and only in that case; if both tables are sharded differently, you can get a wrong result. Another way, number two, is when your first table runs in local mode and the second runs in distributed mode, so on every shard you repeat the same query, select something from t2, and that works, you’ll get a consistent result, but if you have many shards that means for a single query you run tens or hundreds of distributed queries, which may affect your performance. So ClickHouse has a third mode, where it first runs a sub-query or builds a temporary table from the right side of the join on the initiator, then passes that data to all shards, which execute the distributed part and return the result.

So the distributed_product_mode setting shown on the previous slide lets you control the behavior. Normally, if you try to join two distributed tables, you get an exception because it’s denied by default, and the reason it’s denied is that it’s not clear what behavior you want. You can alter it using the local or global settings, which just trigger the different ways to run the query. In principle, you should always run in global mode when queries are rewritten if you have to join distributed queries, but in some rare cases where you have very large tables sharded exactly the same way using exactly the same sharding key, you can run those queries in local mode, and in that case your query performance will be much better than global, because you’ll be joining the pieces of data on every local shard and the end results will be combined.

42:11 — Distributed Query Limitations and Settings

Robert Hodges: Thank you, Alexander. I had Alexander in because this is definitely where it begins to get more complex, and it’s really important to understand that ClickHouse does these distributed queries really fast, but it’s not magic. A few things to be aware of: if a join table is missing, for example if you’re using the defaults the way I was initially, the pushdown will just fail, because it doesn’t know the table is not there. There were some issues with row-level security predicates not being correctly pushed down prior to 20.1, which was recently fixed. There are some minor differences in syntax to get things to propagate correctly; the best policy is to fully qualify your table names, so that as things get pushed down you avoid problems where ClickHouse ends up in the wrong database context and can’t understand where the table is located. And the distributed join behavior is limited; there’s no distributed query optimizer here, so you need to tell ClickHouse what to do and form your queries the right way. If you do that correctly, it’s really fast; if you don’t, you can get errors and sometimes incorrect results. There’s work going on in all these areas to improve this, but that’s the state of things right now.

There are many settings to control distributed queries; I won’t cover them too much because I want to make sure we get to questions, but besides the distributed_product_mode setting, if you’re a MySQL person you know about replica lag, and there’s a setting for maximum delay, as well as load-balancing and prefer-localhost-replica settings. These all let us choose how replicas are chosen, control whether we can skip shards, and so on, so again it goes back to giving ClickHouse hints about how to do the query better.

44:38 — Capacity Planning and What Goes Wrong

Robert Hodges: There are some advanced topics I’d like to go through once your clusters are up and running and you’re putting some load on them. One really important thing for clusters, and ClickHouse in general, is capacity planning. It’s simpler for small data sets, where you often don’t even need it because ClickHouse is so fast and compresses so well that you have a lot of headroom, but as you build larger clusters it’s important to run some tests before deciding on things like sharding keys and host sizes. This procedure is largely thanks to Mik Kocikowski of CloudFlare, who gave a great talk back in December. The way they test is as follows: first, understand the capacity of single nodes, going back to the scaling factors of CPU, RAM, and storage, and in particular load data and understand the local contention between inserts, the background merges that merge your table contents over time, and selects. If you have very high levels of inserts and your parts are too small, for example if you’ve seen “Too many parts” errors, you want to solve that before you go to a cluster, and understand the basic scaling issues like mark cache sizes at the single-node level.

One thing that’s nice is that you may find you don’t need a cluster: if you have a 12- or 24-month design ceiling and a single node will do it, great, just stop there. The one thing to think about is high availability, so you’ll either do backups or introduce replication, but you don’t need sharding. If you do need sharding, you build a cluster and test the full capacity: jam as many inserts as you have on it, see what you need to support the right capacity, and layer on selects, particularly if you have a lot of concurrent selects, where you’ll need to add replicas. The key in this kind of capacity planning is to use your own data; synthetic data does not do the job, because you can’t generate enough of it and it doesn’t have the variability you need to get good results. Slow nodes can be a problem, and in setting up for this talk I had a storage performance problem on Amazon. A great trick is the remote() table function: if you need to see what’s going on across the cluster, you can run the same query across different nodes to see what the performance is, which is a quick way of diagnosing whether one node may have problems. Monitoring systems are really important too, but sometimes you just run against the tables.

There are a number of things that can go wrong in clusters. For very large clusters, ZooKeeper can be a bottleneck if you have huge numbers of parts; that’s not a big problem for smaller installations, but systems like CloudFlare, running 120-plus nodes, hit it. Choosing a bad partition key is hard to get out of, because you discover it’s bad when you usually have a lot of data, so you have to move it around and ClickHouse doesn’t do that magically, so at some point you may have to redistribute your data manually. Systems degrade, both temporarily and permanently, so be on watch for that, and related to all of these, not having enough monitoring will just kill you. In setting up for this talk I had a huge amount of monitoring installed just so I could see what’s going on across the cluster; you cannot be logging in and looking at logs, you need graphs and visualizations and to see data over time.

49:34 — Key Takeaways

Robert Hodges: Let me wrap up quickly. Shards add IOPS, so think of them as adding read and write capacity over a data set; replicas enable more concurrent reads. For sharding keys and clustering patterns, ClickHouse lets you do it any way you want, so you need to think carefully about the right design for your application, and the choices will almost always be dominated by what your application needs to do. Inserting directly to shards over time gets you the best performance, as opposed to going through a distributed table, though many people use distributed tables and are perfectly happy. The distributed query behavior is definitely more complex than running on a local MergeTree, with a number of settings, and you really have to tell ClickHouse what you want to get the right results. The last thing is that when you build a cluster, you’re building a big distributed system, and stuff is going to go wrong, so plan for it: have lots of replicas so if one goes bad you have a replacement, have lots of monitoring, and test on real data so you’re not surprised by obvious problems. One of the key lessons is that if you do clusters well, they’re really fast, so if you’re not seeing fast results, for example if a distributed table isn’t twice or four or eight times faster than a local table, something is wrong, and you should check your setup, because typically the problem isn’t that ClickHouse is doing a bad job, it’s that something is wrong with your resources or configuration.

Speaking of resources, there are lots of other things to read: the Secrets of ClickHouse Query Performance webinar I mentioned, our other webinars including replication, and the talk by Mik Kocikowski, which is a really wonderful description of how to do capacity planning and scale, so definitely check that one out. With that, I’d like to launch the poll; please fill out those three questions, they’ll take about 15 seconds. If anybody wants more information or to consult on their cluster, we’re happy to do a free consultation; the link is below, or send info to info@altinity.com, and we’d be delighted to help. With that, we’ll turn it over to questions.

52:56 — Q&A

Robert Hodges: Alexander, you answered a number of questions in the box; are there any you’d like to go back to verbally?

Alexander Zaitsev: I tried to answer the questions, so maybe we can ask the audience if there’s anything they’d like more detail on.

Robert Hodges: We’ve got a question: how do you remove unreachable instances? This is actually kind of tricky, as sooner or later you may need to go clean up some data in ZooKeeper. This doesn’t happen too often, but if you need to remove a node, you have to get rid of the znode that represents it. There’s work going on to handle that; cleaning up things in ZooKeeper is much easier if the replicas are still alive, which is fairly reliable. There was also a question about an error message regarding replication. Alexander, did you want to take that?

Alexander Zaitsev: That error message says that ClickHouse couldn’t fetch some data from one replica to another, but there is no data loss. It may indicate some network issues or temporary problems, but it’s nothing to be worried about. You need to look into system tables like system.replication_queue and ZooKeeper exceptions in order to set up proper monitoring, and only if something goes really wrong, where you see counters increasing, you may dip into the logs to see what’s actually going on.

Robert Hodges: I got another question: SSD versus HDD, will the query performance be worth the cost? I can answer that. Absolutely, SSDs have a huge impact, particularly where you have a lot of concurrency or otherwise need query results back very quickly. There’s a different set of issues with SSD versus HDD: a big problem people have had on scaling is that they had to pick one or the other, not both, so if you wanted cheap, high-density data with long retention you had to go HDD, and if you wanted very high query speed and concurrency, SSD is better. If you’re thinking about that trade-off, one thing you should definitely look at is tiered storage, which lets you put your hot data on SSD and your cold data on HDD; that’s available now and getting a lot more stable in version 20.

Alexander Zaitsev: If a replica node is temporarily switched off, this is certainly a scenario you may face in production, and there’s nothing to worry about: as soon as the replica is up, ClickHouse will make sure the data has been replicated back.

Robert Hodges: Another question: will a ClickHouse cluster benefit over a single node for an OLAP table with 5,000 updates per second? If you mean updates where you’re actually changing the table, that’s different from inserts; with inserts that’s not a very high number.

Alexander Zaitsev: You can do updates, but one thing you’ll see is that they need to replicate across, and updates and deletes are asynchronous, coming in the background, and in replicated setups they have to be applied on all replicas.

Robert Hodges: There’s actually not a huge difference in performance, Alexander, correct me if I’m wrong, because the basic cost is to go rewrite the data, so if you’re doing an ALTER TABLE UPDATE you’ll have that cost no matter what, and since with replication it basically happens in parallel, there’s not a huge difference in the impact.

Alexander Zaitsev: Yes, you can update a massive amount of data in ClickHouse, but you cannot do a lot of single-point updates like you would in Cassandra or in MongoDB; that’s just not going to work with ClickHouse, because it’s an OLAP system designed to operate with batches of data, not single rows.

Robert Hodges: We had a question: can you revisit the use of remote() to see individual ClickHouse server response times? There it is. Another thing about remote() is that it’s a really handy function; you can put more than one host name in it, so it will effectively create a dynamic, on-the-fly distributed table. So with a comma you could have clickhouse1, clickhouse2, and you’d see the results from both at once. It’s a very handy way of seeing what’s going on across a bunch of servers at once.

Robert Hodges: What tools do you recommend for cluster monitoring? Our standard stack is Prometheus plus Grafana; we use that a lot, and it’s very popular in the ClickHouse community. Beyond that, you can use Zabbix or other monitoring solutions; the key thing is what you’re comfortable with and what gives you the completest picture. For monitoring, it’s important, particularly in these clusters, to not just monitor ClickHouse itself; you need to monitor the whole node, so whatever gets you that is probably your best monitoring solution.

Robert Hodges: Will replicated data be affected in terms of data loss if ZooKeeper goes down, and do we need to back up the ZooKeeper disk? Alexander, do you want to answer this one? It’s a sad topic.

Alexander Zaitsev: Basically, you need to keep ZooKeeper safe. If you lose ZooKeeper, it’s possible but requires quite a lot of manual work to restore a cluster to a working state. If ZooKeeper is down, ClickHouse goes read-only: you can still run queries but cannot insert data. Once ZooKeeper recovers, everything should recover, but if you lose any data in ZooKeeper, you need to recreate it, which means manually reattaching all ClickHouse tables and making sure they’re replicated and the data in ZooKeeper is restored. Just so you understand what ZooKeeper keeps for ClickHouse: it keeps all the metadata about all the data parts and files stored in the file system, and this metadata is used to synchronize replicas between each other. ZooKeeper doesn’t keep any data, only the metadata. At the same time, we don’t recommend backing up ZooKeeper, because you cannot make it consistent with your data anyway, so if you back up ZooKeeper and then restore it, it’s already not the same data you have in your ClickHouse file system, so there’s very little reason to do that. Instead, you should concentrate on making it safe; if you have three ZooKeeper nodes, it’s normally very hard to lose it, and you can also put observers on other sites, which we won’t go into, but there are definitely DR patterns you can follow.

Robert Hodges: There’s a question: is the Kafka ClickHouse engine stable in the latest ClickHouse version?

Alexander Zaitsev: I use the latest stable, 19.16, which at the moment is the most stable, and Kafka works reliably there. In fact, we’re preparing an Altinity blog post on Kafka frequently asked questions, so check it out from time to time, it might answer a lot of questions.

Robert Hodges: We’re over time, so we’ll close up. Someone noted the poll didn’t ask for future topics, which was a surprise to me as well, so if you have things you’d like to see, feel free to send us email at info@altinity.com. We hope you enjoyed this webinar; I learned a lot doing it, and Alexander, thank you so much for your help today. We can handle one more: what about any duplicate or data corruption that can occur during inserts when there’s latency on the shard, replica, or ZooKeeper side?

Alexander Zaitsev: This is a very interesting topic. ClickHouse has support for data deduplication, and it’s based on ZooKeeper. It works as follows: if you insert exactly the same data twice, ClickHouse just silently ignores the second insert. The rationale is that sometimes, if you insert data and have some network problem, you cannot acknowledge to your client that the data was actually inserted, and for reliable insertion the typical approach is to retry on the client side. But when you go to a network error, you do not know whether the data was inserted or not, so retrying could insert the data twice. This only works if you insert into a replicated table and insert exactly the same data, and ClickHouse keeps, by default, the last 100 blocks, so you can make sure you retry after a certain amount of time and it will still work.

Robert Hodges: Great. I think we’ll call it a webinar. Thank you so much everybody; definitely send us questions at info@altinity.com if you’d like to follow up, and we’d be delighted to answer them. We’ll see you at the next webinar. Thank you very much, have a great day.

FAQ

What is the difference between sharding and replication in ClickHouse? Sharding divides a data set into separate pieces spread across nodes, which adds IOPS and therefore read and write capacity over that data set, much like RAID aggregates the performance of multiple disks. Replication keeps copies of the same data on multiple nodes, which doesn’t speed up an individual query but lets many concurrent users read without treading on each other, and also supports high availability. Both are configured at the table level, and a common pattern for large fact tables is sharded and replicated, combining both.

How does ClickHouse replication use ZooKeeper? ClickHouse replication is multi-master and asynchronous. When you insert data into a ReplicatedMergeTree table on any replica, a record is made in ZooKeeper listing the parts that need to be replicated, and the other replicas notice it, contact the original server, and pull the part across. ZooKeeper stores only the metadata about parts and files, not the data itself, and that metadata is what synchronizes replicas. You can make inserts synchronous or set a quorum if you need confirmation that data reached more than one replica.

Should I insert through a distributed table or directly to the shards? Both work. Inserting through a distributed table is convenient because the sharding function decides where data goes, but it’s slower and more resource-intensive because the data is buffered and rewritten, and the default asynchronous behavior carries some risk of data loss if a shard crashes. Inserting directly to the underlying shard tables generally gives the best performance, since shards are just ordinary tables, but it requires your data pipeline to be shard-aware. Many users still use distributed-table inserts and are perfectly happy with them.

How do I choose a sharding key? There are two main approaches. A random key, such as a cityHash of a high-cardinality column, spreads data evenly but means most queries must touch every shard. Keyed sharding, such as by tenant ID, lets you skip shards but tends to produce unbalanced nodes that are hard to control. A bi-level scheme combines both: shard by a top-level key like tenant ID to form a group, then distribute randomly within the group, which limits how many servers a query touches while keeping data balanced. Choose carefully upfront, because ClickHouse does not automatically rebalance data.

How do distributed queries execute across a cluster? A query enters through a distributed table on an initiator node, which pushes as much work as possible down to the shards, where partial aggregate states are computed and then returned to the initiator to be merged. Pushing the WHERE clause and heavy grouping into the innermost query yields close to linear speedups, until queries become fast enough that network overhead dominates. Joins are pushed down by default, and joining two distributed tables is controlled by distributed_product_mode, where global is the safe default and local is faster only when both tables are sharded identically on the same key.

What commonly goes wrong in ClickHouse clusters, and how should I plan capacity? Common issues include ZooKeeper becoming a bottleneck with huge numbers of parts on very large clusters, choosing a bad partition or sharding key that is hard to change once you have a lot of data, gradual or sudden node degradation, and, most damaging, insufficient monitoring. For capacity planning, first max out a single node to understand the contention between inserts, background merges, and selects, then build a cluster and test full capacity using your own real data rather than synthetic data, adding replicas for concurrent selects. Because the slowest node determines distributed query speed, keep the number of nodes a query touches reasonable and monitor the whole node, not just ClickHouse.

© 2026 Altinity, Inc. All rights reserved. Altinity®, Altinity.Cloud®, and Altinity Stable® are registered trademarks of Altinity, Inc. ClickHouse® is a registered trademark of ClickHouse, Inc. Altinity is not affiliated with or associated with ClickHouse, Inc. Kubernetes, MySQL, and PostgreSQL are trademarks and property of their respective owners.