Всем привет, меня зовут Алексей Леонтьев, я техлид и архитектор на проекте Smartbot в компании KTS. Для контекста, Smartbot — это nocode-платформа для настройки сценариев чат-ботов, которые могут работать в нескольких соцсетях и мессенджерах одновременно.
Мы храним информацию о событиях, которые обрабатывает наша платформа. Подробнее о них я расскажу ниже, сейчас важно указать, что все эти события формируют статистику, доступную пользователям. И данные для этой статистики, собранные за год, уже занимают у нас под 2 ТБ дискового пространства, что дорого само по себе, не говоря уже о том, что некоторые запросы в базу стали выполняться заметно дольше ожидаемого.
Раньше для хранения этих данных мы использовали Postgres и разбивали их на партиции по месяцам. Да, можно было бы выгадать какой-то объем за счет оптимизации индексов, но принципиально это не изменило бы картину и привело бы к компромиссам в скорости запросов. Посовещавшись с коллегами, мы приняли решение перенести эти события из Postgres в ClickHouse, так как он умеет эффективно сжимать данные и в целом создан как раз для таких задач: хранение временных рядов и аналитические запросы к ним.
Ниже в статье я расскажу о том, как мы бесшовно перенесли данные из Postgres в ClickHouse. И при этом сократили оперативную память в 4 раза, а размер потребляемого дискового пространства — на порядок.
Оглавление
Еще немного контекста
Чтобы было понятнее, откуда берется эта статистика, немного углубимся в то, как устроен поток событий в Smartbot.
Когда к нам приходит сообщение из соцсети или мессенджера, движок выбирает подходящий сценарий и запускает цепочку действий (блоков) в нем. Аналогичным образом обрабатываются и внутренние события — таймеры и рассылки‑сценарии. По сути, каждое действие сценария оставляет запись о событии.
В моменты пиковой нагрузки таких событий может генерироваться до нескольких тысяч в секунду, и это быстро превращается в очень ощутимый объем данных. При этом пользователи могут задавать разные фильтры для запросов: по времени, каналам, типам событий и блоков и другим критериям. По мере роста продукта объем и нагрузка начали упираться в стоимость хранения и в предсказуемость некоторых запросов — отсюда и возникла задача переезда с Postgres на более подходящее аналитическое хранилище.
Постановка задачи
Основную цель мы сформулировали так: нужно сократить расходы на кластер за счет экономии на дисковом пространстве и памяти так, чтобы не пострадал пользовательский опыт. Эту задачу нам предстояло выполнять в следующих условиях:
Много данных (около 2 ТБ).
Частая вставка большого количества новых строк (до десятков тысяч в секунду).
Редкие запросы на чтение (~20 в час), однако у каждого запроса могут быть свои фильтры по любой колонке.
Ещё более редкие запросы на удаление (от единиц до десятков в сутки), причем удаление сразу по проекту или каналу, не по отдельным строкам.
В итоге мы остановились на таком решении:
Заменить Postgres на ClickHouse.
Для бесшовного переезда использовать связку PeerDB + Temporal.
Описание данных
Давайте более детально взглянем на переносимые данные. В первую очередь речь идет про две большие партицированные таблицы:
stats_labeled_event — события вроде прихода новых сообщений, срабатывания таймеров и прохождения произвольных пользовательских меток, указанных в сценариях ботов.
stats_block_event — то же самое, но по каждому блоку в графе сценария.
spro=> \d+ stats_block_event_part_01_25
Column | Type | Collation | Nullable |
---------------+----------+-----------+----------+
block_kind_id | smallint | | not null |
project_id | bytea | | not null |
user_id | bytea | | not null |
block_id | bytea | | not null |
date_created | date | | not null |
hour_created | smallint | | not null |
channel_id | bytea | | |
Partition of: stats_block_event_parent FOR VALUES FROM ('2024-12-19') TO ('2025-01-19')
Partition constraint: ((date_created IS NOT NULL) AND (date_created >= '2024-12-19'::date) AND (date_created < '2025-01-19'::date))
Indexes:
"stats_block_event_part_01_25_cover_idx" btree (project_id, channel_id, date_created, hour_created, block_kind_id, block_id, user_id)
"stats_block_event_part_01_25_date_created_pid_idx" btree (date_created, project_id)
Access method: heap
------
spro=> \d+ stats_labeled_event_part_01_25;
Column | Type | Collation | Nullable |
--------------+----------+-----------+----------+
project_id | bytea | | not null |
user_id | bytea | | not null |
label_id | integer | | not null |
date_created | date | | not null |
hour_created | smallint | | not null |
channel_id | bytea | | |
Partition of: stats_labeled_event_parent FOR VALUES FROM ('2024-12-19') TO ('2025-01-19')
Partition constraint: ((date_created IS NOT NULL) AND (date_created >= '2024-12-19'::date) AND (date_created < '2025-01-19'::date))
Indexes:
"stats_labeled_event_part_01_25_cover_idx" btree (project_id, label_id, date_created, hour_created, channel_id, user_id)
"stats_labeled_event_part_01_25_pid_ch_id_idx" btree (project_id, channel_id)
Access method: heap
Используемый стек технологий
ClickHouse
ClickHouse — колоночная аналитическая СУБД с открытым кодом, позволяющая выполнять аналитические запросы в реальном времени на структурированных больших данных.
ClickHouse позволяет в явном виде указать, как хранить данные на диске, с помощью ordering key.
Например, во всех пользовательских запросах у нас есть фильтрация по столбцу project_id, а значит, логично указать его в начале ordering key — тогда все данные по этому проекту будут храниться последовательно в указанном порядке. Это в том числе позволяет не хранить на диске дублирующую информацию за счёт алгоритмов сжатия.
По умолчанию ClickHouse использует для сжатия алгоритм zstd — комбинацию сопоставления по словарю (LZ77) и код Хаффмана, но при необходимости можно выбрать и другой алгоритм.
Партицирование по месяцам позволяет сэкономить время при обработке запросов — не загружать в память те файлы, которые заведомо не относятся к указанному в запросе интервалу времени: в большинстве запросов пользователи смотрят статистику за неделю или месяц.
Кроме того, даже партиции в ClickHouse хранятся не целиком, а частями (parts).
Каждый новый INSERT создаёт новый part, и со временем маленькие part'ы объединяются (merge) в более крупные. Ниже мы еще вернемся к нюансам данного процесса.
Данные в ClickHouse
В итоге мы остановились на такой схеме для двух исходных таблиц:
CREATE TABLE stats_block_event ON CLUSTER '{cluster}'
(
`id` Int64,
`project_id` FixedString(12),
`channel_id` FixedString(12),
`user_id` FixedString(12),
`block_kind_id` Int16,
`block_id` FixedString(12),
`date_created` Date32,
`hour_created` Int16,
`_peerdb_synced_at` DateTime64(9) DEFAULT now64(),
`_peerdb_is_deleted` Int8,
`_peerdb_version` Int64
)
ENGINE = ReplicatedReplacingMergeTree('/clickhouse/tables/stats_block_event', '{replica}', _peerdb_version)
PARTITION BY toYYYYMM(date_created)
PRIMARY KEY (project_id, block_id, date_created, hour_created, id)
ORDER BY (project_id, block_id, date_created, hour_created, id)
TTL date_created + INTERVAL 1 YEAR
SETTINGS allow_nullable_key = 1, index_granularity = 8192;
CREATE TABLE stats_labeled_event ON CLUSTER '{cluster}'
(
`id` Int64,
`project_id` FixedString(12),
`channel_id` FixedString(12),
`user_id` FixedString(12),
`label_id` Int32,
`date_created` Date32,
`hour_created` Int16,
`_peerdb_synced_at` DateTime64(9) DEFAULT now64(),
`_peerdb_is_deleted` Int8,
`_peerdb_version` Int64
)
ENGINE = ReplicatedReplacingMergeTree('/clickhouse/tables/stats_labeled_event', '{replica}', _peerdb_version)
PARTITION BY toYYYYMM(date_created)
PRIMARY KEY (project_id, label_id, date_created, hour_created, id)
ORDER BY (project_id, label_id, date_created, hour_created, id)
TTL date_created + INTERVAL 1 YEAR
SETTINGS allow_nullable_key = 1, index_granularity = 8192;
Вообще, PeerDB сам обычно генерирует все необходимые таблицы. Но если вы хотите как-то кастомизировать схемы, проще будет создать все вручную, а потом просто указать те же имена таблиц в интерфейсе. Главное, чтобы при этом в таблицах были «технические» поля, которые нужны самому PeerDB — с одноименным префиксом.
В частности, у нас кастомизированы:
String->FixedString(12)(по факту там хранятсяObjectID, прямо в виде байтов, а не символов).ENGINE(для репликации).PARTITION BY, ORDER BY, PRIMARY KEY, TTL
Репликация в ClickHouse
По умолчанию таблицы создаются на том сервере, на котором был выполнен запрос CREATE TABLE. Чтобы таблица создалась на всех репликах кластера, надо использовать конструкцию ON CLUSTER, а также вместо обычного движка ReplacingMergeTree использовать ReplicatedReplacingMergeTree. После этого писать/читать можно с любой реплики.
Никаких гарантий уникальности ClickHouse не дает. То есть наличие (id) в PRIMARY KEY не гарантирует, что не будет второй записи с тем же id. Однако две записи с одинаковым ordering key со временем будут схлопнуты движком ReplacingMergeTree — останется только та, у которой выше _peerdb_version. В нашем кейсе таких ситуаций с дублированием нет, так как в текущем состоянии платформы мы никогда не изменяем строки, а удаление работает через другой механизм (не через простановку _peerdb_is_deleted).
PeerDB
PeerDB подключается к Postgres и запускает процесс логической репликации. События репликации (создание, редактирование, удаление) строк в Postgres преобразуются в запросы к ClickHouse. Такой подход называется CDC — Change Data Capture.
В нашем случае CDC позволяет бесшовно (без даунтайма) перейти с одного хранилища на другое. Весь процесс состоит из 2 этапов:
Initial Copy — уже существующие данные через S3 переносятся в целевое хранилище.
CDC — новые изменения переносятся из исходной СУБД в целевую (может работать неограниченно долго).
Для оркестрации этого процесса используется Temporal
Разработка
Бэкенд у нас написан на Python 3.11, asyncio. Поэтому для ClickHouse мы решили использовать asynch и clickhouse-sqlalchemy.
Тестирование гипотезы и возникшие проблемы
Для тестирования решения мы взяли из прода по снапшоту одной партиции обеих таблиц. В Postgres они занимали 108 GB, в ClickHouse — 3.4 GB.
Таким образом мы добились сжатия в 31 раз. По производительности запросов ClickHouse обошел Postgres примерно на порядок.
Но без шероховатостей, как водится, не обошлось. Поговорим и о них.
Отсутствие id у строк
Когда-то давно ради экономии места в Postgres мы удалили столбец id. Он просто был не нужен, так как не использовался ни в каких запросах. Но при тестировании переноса данных оказалось, что у нас не сходится количество записей до и после.
Проблема, вероятно, была в механизме дедупликации ReplacingMergeTree. Убрав id, мы оставили записи без уникального ключа, и ClickHouse просто схлопывал те записи, у которых совпадали поля в ordering key. Было решено вернуть и заполнить столбец id.
Скрипт для этого сгенерировал ChatGPT:
id_gen.sh:
#!/usr/bin/env bash
set -euo pipefail
PG_OPTS="host=..."
PART="public.stats_$1_event_part_$2" # run per partition (one or more workers can run this same script)
SEQ="stats_$1_event_id_seq"
BATCH=200000
SLEEP_SECONDS=0.1
while true; do
rows=$(psql "$PG_OPTS" -t -A -c "
WITH c AS (
SELECT ctid FROM $PART
WHERE id IS NULL
FOR UPDATE SKIP LOCKED
LIMIT $BATCH
),
u AS (
UPDATE $PART t
SET id = nextval('$SEQ')
FROM c
WHERE t.ctid = c.ctid
RETURNING 1
)
SELECT count(*)::text FROM u;
" )
rows=$(echo "$rows" | tr -d '[:space:]')
echo "$(date +'%F %T') $PART updated: $rows"
if [ "$rows" = "0" ] || [ -z "$rows" ]; then
echo "done"
break
fi
sleep $SLEEP_SECONDS
done
И все бы ничего, но если добавление новой колонки в Postgres проходит практически мгновенно и не отнимает места на диске, то вот заполнение нового поля в каждой строке буквально удваивает потребляемое место, потому что Postgres создает новые таплы. В итоге пришлось после заполнения каждой следующей партиции делать VACUUM FULL на ней, чтобы вернуть свободное место для продолжения процесса.
Проблемы с CREATE PUBLICATION
Мы переносили данные из Managed Postgres (Selectel). И там недостаточно прав для выполнения:
CREATE PUBLICATION peerdb_publication FOR ALL TABLES;
Пришлось явно перечислять таблицы для переноса (parent'ы):
CREATE PUBLICATION peerdb_publication FOR TABLE stats_block_event_parent, stats_labeled_event_parent, ...;
Проблема в том, что PeerDB при этом отображает список родительских таблиц в интерфейсе, а в публикацию по факту добавляются дочерние таблицы, не родительские:
select * from pg_publication_tables;
pubname | schemaname | tablename
--------------------+------------+--------------------------------
peerdb_publication | public | stats_labeled_event_part_11_24
peerdb_publication | public | stats_labeled_event_part_12_24
peerdb_publication | public | stats_labeled_event_part_01_25
...
В итоге, с помощью нашего DevOps-юнита, мы в несколько подходов пропатчили PeerDB, чтобы он при валидации зеркала сверял родительские таблицы. Но на этом проблемы не закончились.
Пока идет Initial Copy, WAL растет
Как только запускается Initial Copy, WAL перестает чиститься, и место на диске стремительно заканчивается. Это сделано для того, чтобы не были потеряны данные, которые создаются пользователями/ботами с момента запуска Initial Copy и до запуска CDC. Но так как данных очень много, это занимает слишком много времени и места на диске, поэтому такой подход нам не подошел.
Решением стал очередной патч — мы сделали так, чтобы в интерфейсе PeerDB отображались отдельные партиции. Они копируются быстро, да и новые записи добавляются только в последнюю партицию.
Удаление записей в ClickHouse
В ClickHouse есть несколько способов удаления записей:
clickhouse-sqlalchemy генерирует по умолчанию конструкцию ALTER TABLE x DELETE ....
И все бы хорошо, но иногда (не каждый день) мы сталкивались с проблемой, что нагрузка на CPU упирается в потолок, а за ней стремительными темпами выжирается весь диск. Кластер переходит, по сути, в read only: сыпет ошибками Cannot reserve 1.00 MiB, not enough space на все INSERT'ы. Причем минут через 40 ситуация всегда сама приходила в норму.
Мы долго пытались решить проблему тюнингом параметров ClickHouse, но далеко не все там можно изменить, потому что кластер у нас в Yandex Cloud — Managed. В итоге по таймингам и логам удалось понять, что причина была в этих самых ALTER TABLE x DELETE: они приводили к массе мёрджей измененных частей (parts) таблиц.
Пришлось подкостылить, чтобы использовался «обычный» DELETE FROM x.
Он не удаляет строки мгновенно, но проставляет флаги в маске, чтобы записи, помеченные как удаленные, не попадали в выдачу SELECT'ов. Удаление же происходит отложено, в момент очередного мерджа частей таблиц.
Доработки на стороне разработки
Вставка батчами
Из-за описанной выше особенности (что на каждый INSERT ClickHouse создает новые part’ы) и рекомендаций вставлять записи крупными пачками мы пишем события не напрямую в ClickHouse, а через очередь на базе Tarantool и xqueue. О таком применении Tarantool подробно рассказывал мой коллега в своей статье.
Дальше наш отдельный сервис выбирает из этой очереди пачку событий, и одним запросом вставляет их в ClickHouse. Частота вставки зависит от нагрузки, но не реже, чем раз в минуту.
unhex и bytes_as_strings
asynch.Pool(
...
settings={
'strings_as_bytes': True,
},
)
Без этого asynch пытается представить те самые String/FixedString в виде utf8-строк и падает, ведь там на самом деле произвольные байты.
Фильтрация по "байтовым" полям
Чтобы использовать такие строки, состоящие из байтов в запросах, нужно пропатчить код генерации запросов. А именно — обернуть параметры в unhex().
def compile_query(self, query: Select | UpdateBase | str) -> tuple[str, dict]:
compiled = query.compile(dialect=clickhouse_dialect)
sql_text = compiled.string
params = dict(compiled.params)
# Walk through bind params and fix FixedString ones
for key in compiled.params.keys():
if isinstance(compiled.binds[key], BindParameter):
col_type = getattr(compiled.binds[key].type, '__visit_name__', None)
if col_type == 'BYTEA':
# Replace placeholder in SQL with unhex(...)
sql_text = sql_text.replace(f'%({key})s', f'unhex(%({key})s)')
# Hex encode the value
value = params[key]
if isinstance(value, list):
params[key] = [
v.hex()
for v in value
]
else:
params[key] = value.hex()
return sql_text, params
Тестирование в продакшене
Конфиг железа в проде стал заметно скромнее.
Было: 6 vCPU, 64 GB RAM, 1.95 TB
Стало: 4 vCPU, 16 GB RAM, 180 GB
Занятое место на диске сократилось в 14 раз: 1659 ГБ удалось ужать до 120.
Пользовательские запросы
Мы не проводили полноценного нагрузочного тестирования. Запросы на чтение этих данных приходят к нам только от пользователей, когда они заходят в раздел Статистика. Если на запись может быть тысячи строк в секунду, то на чтение — десятки запросов в час, так что вопрос производительности при конкурентной нагрузке тут не стоит. Мы ограничились ручными замерами на нескольких типовых запросах, собранных по логам.
Статистика по событию за неделю
--- postgres:
SELECT
count(sle.user_id) AS count_1,
sle.date_created
FROM stats_labeled_event_parent sle
WHERE sle.project_id = '\x6676733f497c7a4a21667c7a'::BYTEA
AND sle.label_id = 1
AND sle.date_created BETWEEN DATE '2024-11-19' AND DATE '2024-11-26'
GROUP BY sle.date_created
ORDER BY sle.date_created;
--- clickhouse:
SELECT
count(user_id) AS count_1,
date_created
FROM stats_labeled_event
WHERE (project_id = unhex('6676733f497c7a4a21667c7a')) AND (label_id = 1) AND ((date_created >= toDate('2024-11-19')) AND (date_created <= toDate('2024-11-26')))
GROUP BY date_created
ORDER BY date_created;
Postgres: 8.6 мсек
ClickHouse: 24 мсек
Тот же запрос, но за месяц, показал следующие результаты:
Postgres: 8 мсек
ClickHouse:13 мсек
Уникальные пользователи, фильтрация по типу блока
--- postgres
SELECT
count(distinct sle.user_id) AS count_1,
sle.date_created,
sle.hour_created
FROM stats_block_event_parent sle
WHERE sle.project_id = '\x6676733f497c7a4a21667c7a'::BYTEA
AND sle.block_kind_id = 1
AND sle.date_created BETWEEN DATE '2024-11-19' AND DATE '2024-12-20'
AND (sle.date_created != DATE '2024-11-19' OR sle.hour_created >= 10)
AND (sle.date_created != DATE '2024-12-20' OR sle.hour_created <= 18)
GROUP BY date_created, hour_created
ORDER BY date_created, hour_created;
--- clickhouse
SELECT
countDistinct(user_id) AS count_1,
date_created,
hour_created
FROM stats_block_event
WHERE (project_id = unhex('6676733f497c7a4a21667c7a')) AND (block_kind_id = 1) AND ((date_created >= toDate('2024-11-19')) AND (date_created <= toDate('2024-12-20'))) AND (
(date_created != toDate('2024-11-19')) OR (hour_created >= 10)) AND ((date_created != toDate('2024-12-20')) OR (hour_created <= 18))
GROUP BY
date_created,
hour_created
ORDER BY
date_created ASC,
hour_created ASC;
Postgres: 4442 мсек
Clickhouse: 54 мсек
Тот же запрос с фильтрацией по block_id и за 1 день дал следующие результаты:
Postgres: 1349 мсек
ClickHouse: 951 мсек
Фоновые запросы
Кроме пользовательских запросов мы также обрабатываем фоновые. Например, запрос, который периодически обновляет число событий за последние 7, 30, 60, и 180 дней по каждому проекту. Эта информация используется в разных местах бизнес-логики для оценки активности в проекте.
SELECT
project_id,
COUNT(CASE WHEN date_created > '2024-06-12' THEN '1' END) AS count_7d,
COUNT(CASE WHEN date_created > '2024-05-20' THEN '1' END) AS count_30d,
COUNT(CASE WHEN date_created > '2024-04-20' THEN '1' END) AS count_60d,
COUNT(CASE WHEN date_created > '2023-12-22' THEN '1' END) AS count_180d
FROM
stats_block_event_parent
WHERE
date_created > '2024-06-18'
GROUP BY
project_id;
В Postgres такой запрос порой занимал больше часа, джоба исчерпывала таймаут, иногда запрос падал с ошибкой ERROR: canceling statement due to conflict with recovery. После перехода на ClickHouse запрос стал выполняться 20-25 минут.
Итоги
Поставленная цель была достигнута:
Размер диска удалось сократить в 10 раз.
Потребовалось в 4 раза меньше оперативки.
Время обработки пользовательских запросов стало более стабильным. Почти все запросы стали обрабатываться за 10-50 мсек.
ClickHouse отлично подошел для решения поставленной задачи, а PeerDB обеспечил бесшовный перенос данных.
Надеюсь, материал был вам полезен. С радостью отвечу на возникие вопросы.
Рекомендую статьи моих коллег, которые рассказывают о других способах оптимизации бэкенда:
