ClickHouse is used more and more often, and it has proven itself well for building data warehouses and running analytics projects. Thanks to its column-oriented architecture, ClickHouse combines high insert throughput for large data volumes (millions of rows per second) with incredibly fast full scans over large tables. That speed, however, comes with certain limitations, and users have to keep a number of caveats in mind — caveats that directly affect the design and development of solutions built on ClickHouse.
This article shows the difficulties the team ran into because of ClickHouse's architectural limitations and how they were overcome, cutting the resources consumed by the ClickHouse server by nearly 50×.
The task at hand
ClickHouse receives an online stream of transactions — essentially historical data on how many units of each resource (products/goods) at what cost were moved from points of sale to the warehouse and back. A set of transformations runs over the data constantly flowing into the database, such as custom transaction deduplication, enrichment from reference tables, and aggregation.
The transformations produce a turnover mart with the following structure:
DROP TABLE IF EXISTS dm_turnover;
CREATE TABLE dm_turnover
(
`date` Date,
`pos` String,
`storage` String,
`productCode` UInt32,
`amount` Decimal(38, 3),
`cost` Decimal(38, 6),
`currency` String,
`supplyNumber` UInt64,
`updateTime` DateTime('Europe/Moscow')
)
ENGINE = MergeTree
PARTITION BY toYYYYMMDD(date)
ORDER BY (date, pos, storage, productCode);
The key part of the task is building a daily balance mart — in practice, this is a daily cumulative view built from the turnover mart. DDL for the balance mart:
DROP TABLE IF EXISTS dm_balance;
CREATE TABLE dm_balance
(
`date` Date,
`pos` String,
`storage` String,
`productCode` UInt32,
`amount` Decimal(38, 3),
`cost` Decimal(38, 6),
`currency` String,
`updateTime` DateTime('Europe/Moscow')
)
ENGINE = MergeTree
PARTITION BY toYYYYMMDD(date)
ORDER BY (date, pos, storage, productCode);
As the balance mart's DDL shows, its structure is practically identical to the turnover mart's. The way the balance mart is populated is quite different, though, and that difference creates certain problems, described below.
Data volumes
First, a few words about data volumes. Throughout 2022 the turnover mart collected between 3.6 and 7.8 million rows per day and occupied about 29 GB on disk. Each day's data from the turnover mart is combined with the previous state and aggregated by the key (pos, storage, productCode) into the balance mart.
The balance mart is built as follows (shown for a few dates):
( '2022-12-31' in dm_balance ) = ( '2022-12-31' in dm_turnover ) UNION ( '2022-12-30' in dm_balance )
( '2022-12-30' in dm_balance ) = ( '2022-12-30' in dm_turnover ) UNION ( '2022-12-29' in dm_balance )
( '2022-12-29' in dm_balance ) = ( '2022-12-29' in dm_turnover ) UNION ( '2022-12-28' in dm_balance )
...
As the example shows, the balance mart is assembled from the turnover mart and has a cumulative effect: whatever has been accumulated in the balance up to yesterday's date is merged with the turnover for the current date. Every day.
Following this logic, the number of rows in the balance mart can only grow, as new (pos, storage, productCode) combinations appear in the turnover mart day after day. In 2022 the balance mart collected between 65 and 143 million rows daily, and its weight over the year came to more than 650 GB. Row counts and daily data volumes keep growing with every day.
Problems with building the balance mart
The balance mart has to be kept up to date even though the turnover mart can be appended to or corrected for previous dates. For example, if a batch of correcting transactions arrives for date T-60 (about two months earlier), the turnover mart changes for T-60, and the balance mart has to be rebuilt sequentially for each date, starting at T-60 and running through the current day. Formally, rebuilding the balance mart for a single date is done with the following sequence of queries:
ALTER TABLE tmp_balance DROP PARTITION :partKey; -- (*)
INSERT INTO tmp_balance
WITH :partKey AS partDate
SELECT
toDate(toString(partDate)) date,
pos,
storage,
productCode,
SUM(amount) amount,
SUM(cost) cost,
currency,
MAX(updateTime) updateTime
FROM (
SELECT pos, storage, productCode, amount, cost, currency, updateTime
FROM {CH_SOURCE_TABLE}
WHERE toYYYYMMDD(date) = partDate
UNION ALL
SELECT pos, storage, productCode, amount, cost, currency, updateTime
FROM {CH_TABLE}
WHERE date = toDate(toString(partDate)) - INTERVAL 1 DAY
) GROUP BY pos, storage, productCode, currency;
ALTER TABLE dm_balance DROP PARTITION :partKey;
ALTER TABLE tmp_balance MOVE PARTITION :partKey TO TABLE dm_balance;
* A small remark on why the team uses an intermediate table tmp_balance and moves whole partitions around. This is one more consequence of how ClickHouse is architected — in particular, the impossibility of running classic DML DELETE/UPDATE operations quickly and often. These operations are executed in ClickHouse as DDL (see https://clickhouse.com/docs/en/guides/developer/mutations/). Rebuilding marts by whole partitions is therefore much faster and more efficient: instead of heavyweight, slow UPDATE/DELETE we use fast INSERT and DROP/MOVE PARTITION. For the same reason the team decided not to use the dbt tool — it supports only DML operations, and it turned out to be unsuitable for this task, both in terms of speed and in its inefficient use of the ClickHouse server's resources. In other tasks (smaller volumes, incremental data updates only, relaxed speed requirements) dbt is certainly convenient — just not in this one.
To make the process clearer, here is how the balance mart is refreshed when the turnover mart changes for a previous date.
So the main difficulty the team hit was a lack of memory to run the query. The problem becomes acute when the query contains a subquery. The subquery's result set has to fit in the ClickHouse server's memory and must not exceed max_memory_usage/max_memory_usage_for_user, otherwise the query is aborted with an error DB::Exception: Memory limit (for query) exceeded.
With the balance mart's build logic there is no way to avoid a subquery, since data from two marts has to be merged and grouped.
When merging 143 million rows from the balance mart with 7.8 million rows from the turnover mart, the "INSERT INTO tmp_balance" query (given above) requires up to 55 GB of RAM. Keep in mind that this is not the only query running on the ClickHouse host. The database has other parallel activity related to transforming and fetching data (refreshing the same turnover mart, for example).
So keeping just one balance mart up to date requires a ClickHouse host with ~64 GB of RAM. Yes, you can keep increasing the capacity of the ClickHouse server as volumes grow and the system will keep running correctly, but that cannot be called a rational fix for the problem — it would be more of a temporary workaround.
The team did find a solution that addresses the hardware resource costs without a significant loss of performance and data processing speed.
How it was solved
The solution turned out to be changing the partitioning of the balance mart into smaller pieces and reworking the transformation application. Where one partition used to hold a full day of data (a single date), a day's data is now spread across 50 different partitions that differ from each other by the remainder (mod) of the product code. The final DDL of the balance mart:
DROP TABLE IF EXISTS dm_balance;
CREATE TABLE dm_balance
(
`date` Date,
`pos` String,
`storage` String,
`productCode` UInt32,
`amount` Decimal(38, 3),
`cost` Decimal(38, 6),
`currency` String,
`updateTime` DateTime('Europe/Moscow')
)
ENGINE = MergeTree
PARTITION BY toUInt64(concat(toString(toYYYYMMDD(date)), toString(productCode % 50)))
ORDER BY (date, productCode, pos, storage);
Splitting the balance mart into smaller partitions also made the transformation application's logic more complex. Where the team used to simply rebuild consecutive daily partitions in order, from date T-N to T, refreshing the balance for a given date now has to happen only for the specific small partitions — those where the turnover mart changed.
To make the new flow clearer, here is how the balance mart is refreshed now.
The balance rebuild process now resembles raindrops on a window: individual drops slide down from the very top, picking up other drops along the way :)
Below is the transformation algorithm as code.
The team usually runs transformations on serverless infrastructure (AWS Lambda): it is cheaper and more convenient than maintaining a virtual machine in the cloud or on premises, since only the number of invocations and the GB*hour consumed are billed. This article presents generic code without the details of the infrastructure it runs in, so that you can adapt it to your own needs and infrastructure.
The transformation application implements the following methods:
1) Scheduled start. The application first checks that it is running in Singleton mode (more on how this was done below), and if no other instances of the application are running at the moment, it identifies the partitions in the turnover mart where changes occurred (a discrepancy in update_time against the balance mart). If changes occurred in two partitions with the same %50 modulo, the earliest date is kept. The output of this method is the dates and partition numbers that need to be rebuilt. Another application method is then called with that output. The choice of how to pass data between the application's functions is left to the developer — it can be a direct call from some controller method, or a transfer through a queue, a database, and so on.
2) Single-date transformation. A recursive method that processes the partitions of the earliest date in the batch from the request, rewrites the request — shifting the just-rebuilt partitions to the next date — and passes the call to itself to transform the next date in turn.
First, the auxiliary method and table that help the application work in Singleton mode, regardless of the number of partitions being processed at the moment and the depth of processing. In ClickHouse, create the following table:
DROP TABLE IF EXISTS dm_balance_log;
CREATE TABLE dm_balance_log
(
`parent` UUID,
`child` UUID,
`op` UInt8,
`insertTime` DateTime('Europe/Moscow') DEFAULT now()
)
ENGINE = MergeTree
PARTITION BY toYYYYMMDD(insertTime)
ORDER BY insertTime
TTL toStartOfDay(insertTime) + INTERVAL 3 DAY;
The table serves to store the execution log and at the same time helps the application run in Singleton mode when launched on schedule.
To write to the dm_balance_log table, the following auxiliary method is implemented:
def trail(parentId, childId, op):
query = '''INSERT INTO {CH_TABLE}_log (* EXCEPT insertTime) VALUES ('{parentId}', '{childId}', {op})'''.format(
parentId=parentId, childId=childId, op=op, CH_TABLE=CH_TABLE
)
clickhouse.execute(query)
Next, the method launched periodically on schedule, which searches for changed partitions in the source turnover mart:
import os
from clickhouse_driver import Client
from datetime import datetime, timedelta
import pytz
CH_HOST = os.getenv('CH_HOST')
CH_USER = os.getenv('CH_USER')
CH_PASSWORD = os.getenv('CH_PASSWORD')
CH_DB = os.getenv('CH_DB')
CH_SOURCE_TABLE='dm_turnover'
CH_TABLE='dm_balance'
CH_TMP_TABLE='tmp_balance'
CH_MODULO_BASE=50
clickhouse = Client(host=CH_HOST,
user=CH_USER,
password=CH_PASSWORD,
database=CH_DB,
secure=True)
def changed_partitions(requestId):
query = '''
WITH {CH_MODULO_BASE} AS MODULO_BASE,
(SELECT
parent=child AND op = 1
OR insertTime < now() - INTERVAL 15 MINUTE
FROM {CH_TABLE}_log ORDER BY insertTime DESC LIMIT 1) AS isUnlocked,
rawParts AS (
SELECT concat(toString(toYYYYMMDD(date)), toString(modulo(productCode,MODULO_BASE))) partKey, max(updateTime) maxUpdateTime
FROM {CH_SOURCE_TABLE}
WHERE (isUnlocked = 1 OR isNull(isUnlocked))
AND date >= toDate('20220101')
AND updateTime > toStartOfMinute(now()) - INTERVAL {INTERVAL}
group by partKey),
parts AS (
SELECT _partition_id as partKey, max(updateTime) maxUpdateTime
FROM {CH_TABLE}
WHERE _partition_id IN (SELECT partKey FROM rawParts)
GROUP BY partKey),
changedParts AS (
SELECT substring(partKey, 1, 8) dt, toUInt64(substring(partKey, 9)) modul
FROM rawParts rp LEFT JOIN parts p USING partKey
WHERE rp.maxUpdateTime > p.maxUpdateTime
ORDER BY modul, dt
LIMIT 1 BY modul)
SELECT dt, groupArray(modul) modules FROM changedParts
GROUP BY dt
ORDER BY dt
'''.format(
INTERVAL='1 HOUR',
CH_SOURCE_TABLE=CH_SOURCE_TABLE,
CH_TABLE=CH_TABLE,
CH_MODULO_BASE=CH_MODULO_BASE
)
msg = {
'process': {},
'next': [],
'parentId': requestId
}
cnt = 0
settings = {'max_block_size': 100000}
rows_gen = clickhouse.execute_iter(query, settings=settings)
for row in rows_gen:
dt = row[0]
modules = row[1]
if cnt == 0:
msg['process'] = {'dt': dt, 'modules': modules}
else:
msg['next'].append({'dt': dt, 'modules': modules})
cnt += 1
if cnt > 0:
# invoke "transform" method with msg
...
else:
# nothing to process / skip
...
Then the method that performs the actual transformation for the earliest date in the request, and recursively calls itself when necessary:
# ... the same import and global vars
# ... the same clickhouse client initialization
# ... the implementation of method "changed_partitions"
def transform(msg, requestId):
partDate = msg['process']['dt']
partModules = msg['process']['modules']
parentId = msg['parentId']
trail(parentId=parentId, childId=requestId, op=0)
for p in partModules:
partModule = str(p)
partKey = partDate + partModule
clickhouse.execute('ALTER TABLE {CH_TMP_TABLE} DROP PARTITION {partKey};'.format(
CH_TMP_TABLE='dm_balance', partKey=partKey))
query = '''
INSERT INTO {CH_TMP_TABLE}
WITH {CH_MODULO_BASE} AS MODULO_BASE,
toString({partDate}) AS partDate,
{partModule} AS partModule
SELECT
toDate(partDate) date,
pos,
storage,
productCode,
SUM(amount) amount,
SUM(cost) cost,
currency,
MAX(updateTime) updateTime
FROM (
SELECT pos, storage, productCode, amount, cost, currency, updateTime
FROM {CH_SOURCE_TABLE}
WHERE _partition_id = partDate
AND modulo(productCode, MODULO_BASE) = partModule
UNION ALL
SELECT pos, storage, productCode, amount, cost, currency, updateTime
FROM {CH_TABLE}
WHERE _partition_id = concat(toString(toYYYYMMDD(toDate(partDate) - INTERVAL 1 DAY)), toString(partModule))
) GROUP BY pos, storage, productCode, currency;
'''.format(
CH_TMP_TABLE=CH_TMP_TABLE,
CH_SOURCE_TABLE=CH_SOURCE_TABLE,
CH_TABLE=CH_TABLE,
partDate=partDate,
partModule=partModule,
CH_MODULO_BASE=CH_MODULO_BASE
)
clickhouse.execute(query)
query = 'ALTER TABLE {CH_TABLE} DROP PARTITION {partKey};'.format(
CH_TABLE=CH_TABLE, partKey=partKey
)
clickhouse.execute(query)
query = 'ALTER TABLE {CH_TMP_TABLE} MOVE PARTITION {partKey} TO TABLE {CH_TABLE};'.format(
CH_TABLE=CH_TABLE, CH_TMP_TABLE=CH_TMP_TABLE, partKey=partKey
)
clickhouse.execute(query)
trail(parentId=parentId, childId=requestId, op=1)
nextDate = datetime.strptime(partDate, '%Y%m%d').replace(
tzinfo=pytz.timezone('Europe/Moscow')).date() + timedelta(days=1)
today = datetime.now().replace(
tzinfo=pytz.timezone('Europe/Moscow')).date()
if nextDate <= today:
msg['process']['dt'] = nextDate.strftime('%Y%m%d')
if 'next' in msg and len(msg['next']) > 0 and 'dt' in msg['next'][0] and msg['next'][0]['dt'] == msg['process']['dt']:
msg['process']['modules'] = msg['process']['modules'] + \
msg['next'][0]['modules']
msg['process']['modules'] = sorted(set(msg['process']['modules']))
del msg['next'][0]
# ... invoke transform recursively with msg
else:
trail(parentId=parentId, childId=parentId, op=1)
One caveat: the code above is not a reference implementation for Python developers, and there is certainly room for improvement in it. The team only wanted to formalize and demonstrate the algorithm the application uses to solve this task.
In general, this approach works for building balance marts when you need to collect (and constantly update) in one place the result of a query with a window function (window function). Across an entire table, however, that query does not run — there is not enough memory, given the large volumes in the source tables.
By adjusting the degree of partitioning (in this case — the number of partitions per date), you can control the memory the transformation application's queries consume on the ClickHouse server.
Conclusion
As a result, with a divisor of 50 that splits dates by product code into separate partitions, and with each atomic operation executed sequentially, the team cut memory consumption in ClickHouse by 50×: from 55 GB to 1.1 GB!
The downsides are:
- More complex transformation logic and, as a result, more expensive maintenance and development;
- Rebuilding the balance mart slowed down by roughly 10–20%;
- Some SELECT queries against the balance mart became more complex, because the number of partitions processed by a query grew significantly, and partitions are no longer so clearly tied to time (the
datefield). Adding predicates that carry conditions on partition names to such queries optimizes their execution and restores the previous computation speed.
For each of these downsides there is either an alternative solution, or they are fully offset by the big upside — a many-fold reduction in the resources (RAM) required by the ClickHouse server, and the overall viability of the solution.
The team sees this case as a good illustration of how important it is to think through and design solutions on ClickHouse in detail. Because of the specifics and limitations, a solution/system/application may simply fail to work if the data volumes are not taken into account in time and, accordingly, the approach to processing (transforming) the data is not adapted.
Finally, a small piece of advice for anyone who has to build solutions on ClickHouse: partition large tables sensibly, so that later you can cut the resources spent on data processing and possibly simplify development or shorten the development time. Intuitive partitioning (by date only, for example) will not always let you work with the data normally later on — sometimes it leads to a solution that does not work (as in this example). At the same time, overly detailed partitioning (lots of small partitions) can significantly increase transformation time and complicate the logic of the applications that implement it. So it is important to establish the data volumes being processed and the requirements for how fast they must be prepared as early as possible, and to find a middle ground between the complexity of table partitioning and the logic of transformations.
Good luck!