Всем привет! Меня зовут Мурад Арфанян, я разработчик информационных систем в Ozon Tech. Наша команда работает с данными жизненного цикла товаров в логистике. 

Объём продаж растёт стремительными темпами, и нешардированный ClickHouse уже не справляется с постоянно увеличивающимися потоками данных. Чтобы решить эту задачу, мы построили шардированный кластер, преодолев на пути несколько интересных технических вызовов. В этой статье я расскажу о нашем опыте и решениях, которые помогли масштабировать систему и обеспечить стабильную работу при росте нагрузки.

Содержание

• Старый кластер ClickHouse

• Причина перехода на новый кластер ClickHouse

• Архитектура на новом кластере

• Ключ шардирования и глобальные джойны

• Схема приёма и обработки данных

• Обновление витрин данных

• Важные моменты

• Заключение

Старый кластер ClickHouse

Наш старый ClickHouse был нешардированным кластером. За полтора года работы мы увеличивали мощность сервера, добавляя процессоры, память и дисковое пространство — то есть масштабировались вертикально. На тот момент объём данных в логистике был значительно меньше, и мы рассчитывали, что сможем эффективно контролировать бизнес-ограничения. Однако на практике мы упёрлись в возможности вертикального масштабирования быстрее, чем рассчитывали. 

Причина перехода на новый кластер ClickHouse

Сначала мы тщательно проанализировали прогнозируемый прирост данных и количество критичных процессов, завязанных на команду, а также потенциальные лимиты вертикального масштабирования. Поняли, что скоро упрёмся в ограничения железа и не сможем справиться с резким увеличением нагрузки, связанной с сезонностью и другими факторами, а также не сможем обеспечивать требуемые SLA в поставке данных. После этого мы пообщались с внутренними специалистами из платформенной команды ClickHouse, которые специализируются на поддержке и сопровождении,и пришли к решению перейти к шардированию. Хотя мы понимали, что этот шаг неизбежен, он наступил значительно раньше, чем ожидалось. В итоге мы запустили процесс перехода на новый шардированный кластер.

Основные преимущества использования шардирования:

• Горизонтальное масштабирование.

  Шардирование позволяет масштабироваться горизонтально, то есть добавлять новые узлы для распределения нагрузки и увеличения объёма хранимых данных. Это делает систему более гибкой и позволяет справляться с ростом нагрузки и объёма данных без необходимости дорогостоящего апгрейда оборудования.

• Ускорение обработки запросов.

  Шардирование распределяет данные между узлами на основе ключа шардирования. Ключ шардирования может быть случайным rand() или выводом хэш-функции. Обычно с результатом хэш-функции выполняют определённые вычисления, например, берут остаток от деления этого хэша на количество физических или виртуальных шардов, чтобы получить идентификатор шарда. В результате каждый узел обрабатывает только свою часть данных параллельно с другими, что значительно ускоряет время ответа на запросы и повышает пропускную способность системы.

• Гибкость архитектуры.

  Шардирование реализует архитектуру shared-nothing, когда каждый узел шардов автономен; они не используют совместно одни и те же данные или вычислительные ресурсы. Это упрощает масштабирование и обслуживание отдельных компонентов системы.

Единственной причиной, по которой наша команда приняла это решение, была невозможность дальнейшего вертикального масштабирования в рамках существующей инфраструктуры. Это привело нас к необходимой мере — горизонтальному масштабированию. Без этой ключевой потребности мы продолжали бы использовать нешардированный ClickHouse. Однако переход на новый шардированный ClickHouse привёл к ряду сложностей, о которых мы поговорим ниже.

Архитектура на новом кластере

Ранее в нашей инфраструктуре использовался единственный production-кластер ClickHouse, на котором велись и разработка, и эксплуатация. Отсутствие миграций и управления изменениями в базе данных приводило к случайным некорректным действиям разработчика и риску положить старый production-кластер.

Так как мы решили перейти на новый, чистый кластер и наполнять его с нуля, при этом из-за особенностей шардирования невозможен простой перенос таблиц и запросов, были приняты следующие решения (Рис. 1):

• Создать рядом с production-кластером pre-production. В отличие от традиционной staging-среды, которая часто бывает упрощённой и с ограниченными данными, pre-production у нас таков: 

  • конфигурация сопоставима, но на pre-production CPU и объём памяти меньше, при этом диски одинаковые;

  • позволяет максимально точно воспроизводить реальные сценарии работы;

  •  служит основным местом для проверки миграций и новой функциональности.

    Хотя такой подход не самый оптимальный с точки зрения использования ресурсов, он удобный и надёжный.

• Внедрить систему миграций для управления изменениями в базе данных, которые сначала применяются на pre-production, а затем — на production-среде. Для этого было выбрано платформенное решение, построенное на основе goose migrator. Это позволило удобнее создавать и выполнять миграции между средами разработки.

• Мигрировать на более свежую версию ClickHouse. Мы уже несколько раз обновляли ClickHouse на новых кластерах и на текущий момент остановились на версии 24.8 LTS. В ней исправлены некоторые проблемы с утечками памяти, улучшена работа с FINAL, добавлена поддержка RecursiveQuery, есть и другие важные улучшения. На старом кластере установлена версия 22.8 LTS.

Ключ шардирования и глобальные джойны

Одной из ключевых сложностей при переезде на новый шардированный кластер стала задача выбора подходящей функции хеширования с учётом будущего возможного решардинга. Хотя в большинстве случаев достаточно использовать простую функцию rand(), нам было важно обеспечить локальные джойны — то есть сделать так, чтобы связанные данные находились на одном шарде для минимизации затрат на глобальные соединения (global join).

Переход на шардированный кластер позволил использовать движок Distributed для маршрутизации запросов между узлами, где каждый шард содержит уникальную часть данных. После анализа различных вариантов мы выбрали функцию хеширования JumpConsistentHash — быстрый и экономичный по памяти алгоритм консистентного хеширования. 

В качестве основного ключа для большинства таблиц изначально был выбран идентификатор сортировочного центра. Однако на практике оказалось, что этот ключ редко применяется в локальных джойнах и фильтрациях, а также приводит к неравномерному распределению данных между шардами. В результате было принято решение изменить схему шардирования, выбрав в качестве ключа номер посылки, что позволило улучшить балансировку данных на шардах и повысить эффективность запросов, используя локальные джойны.

Теперь немного о глобальных джойнах (GLOBAL JOIN) в ClickHouse. Они применяются при работе с распределёнными таблицами и отличаются от обычных джойнов способом обработки подзапросов и передачи данных между серверами кластера. При обычном джойне подзапросы выполняются на каждом сервере кластера отдельно, и правая таблица для соединения формируется локально на каждом сервере. В случае глобальных джойнов подзапрос сначала выполняется на сервере-инициаторе, результат сохраняется во временную таблицу и затем передаётся на все удалённые серверы, где уже выполняется основной запрос с использованием этих данных. Такой подход позволяет избежать избыточных вычислений и сетевого трафика.

На практике при миграции на новый кластер мы придерживались простого правила: если возможно выполнить локальный JOIN или IN — используем их, в противном случае применяем GLOBAL JOIN или GLOBAL IN. Тем не менее, при миграции и последующей работе на новом кластере мы обнаружили ряд неочевидных нюансов, которые заслуживают отдельной статьи.

Схема приёма и обработки данных

Схема приёма и обработки данных из Kafka при переходе на новый кластер изменилась незначительно, добавился шаг записи в Distributed-таблицу (Рис. 2):

• Чтение из топика Kafka: данные поступают из внешнего инсертера к нам батчами сообщений из топика Kafka. Принимаем в двух форматах: protobuf и json.

• Запись в таблицу kafka_* с Engine = Null: полученные батчи сообщений записываются в таблицу ClickHouse с префиксом kafka_. Эта таблица использует движок Engine = Null, что означает — она не хранит данные, а служит промежуточным приёмником для дальнейшей обработки.

• Материализованное представление (MV) в Distributed-таблицу: используется у нас Materialized View для преобразования и агрегации данных из таблицы kafka_*. Обработанные данные (преимущественно сложной структуры) записываем в Distributed-таблицы, которые распределяют данные по узлам кластера, а сами распределённые данные хранятся в таблицах с постфиксом *_local.

Обновление витрин данных

Ещё один важный аспект, косвенно связанный с шардированием и который нам пришлось пересмотреть при миграции, — это поддержание целостности данных при обновлении витрин данных. Эта тема обширная и также заслуживает отдельной статьи.

В результате экспериментов и анализа были выделены следующие варианты, из которых два последних активно применяются в нашем шардированном кластере:

1. ReplacingMergeTree — простой и понятный механизм. У нас есть таблица с данными версии ver1, которую нужно заменить или обновить данными версии ver2. Этот подход эффективен в случаях, когда не требуется удаление старых данных, а нужно только дообогащение таблицы актуальными данными. 

2. EXCHANGE TABLES — метод предполагает быструю замену содержимого двух таблиц. Команда воздействует только на метаданные таблиц — атомарно отсоединяет и присоединяет таблицы с обновлёнными метаданными. 

3. ALTER TABLE ... REPLACE PARTITION — используется для копирования партиции из одной таблицы в другую с заменой существующей партиции в целевой витрине. 

4. View + DAG timestamp + timestamp в индексе — для обновления витрин каждый DAG фиксирует время начала и завершения своей работы в таблице метаданных, создавая таким образом уникальную «версию транзакции», подтверждающую успешное выполнение. Для вывода актуальных данных используется View, которая фильтрует записи по временной метке последнего коммита.

5. Версионирование по партициям — DAG обращается к таблице метаданных, чтобы узнать активную партицию, затем удаляет противоположную партицию, загружает новые данные под активной партицией и обновляет запись в таблице метаданных, делая её активной. Финальная витрина через View всегда выбирает данные только из текущей активной партиции.

Важные моменты

Ниже приведу некоторые из наших наблюдений и подходов, выделенных в работе с ClickHouse.

• Одна из самых распространённых (хотя и очевидных) проблем при переходе на шардирование — при использовании локальных таблиц с движком Replacing обязательно включайте ключ дедупликации в ORDER BY. Replacing определяет дубликаты по значениям колонок, указанных в ORDER BY. Если ключ дедупликации не включён в ORDER BY, ClickHouse не сможет корректно определить уникальные строки, и при использовании FINAL на Distributed-таблице результат может быть некорректным — останутся лишние дубликаты или будут потеряны актуальные данные. 

• В конце месяца мы сталкиваемся с сокращением свободного места на диске, поскольку TTL настроен на месячный интервал. Это приводит к тому, что удаление устаревших данных происходит пакетно — только когда срок действия TTL истекает для целых партиций или частей данных. Мы экспериментировали с разными вариантами TTL и схемами партиционирования: месячные, недельные, дневные. На практике, если вы храните данные всего 1–2 месяца, оптимально использовать более мелкие партиции — по неделям или дням. Это позволяет удалять устаревшие данные более равномерно и эффективно, а не ждать конца месяца, когда удалится большой объём данных сразу. 

  Также важно учитывать настройку ttl_only_drop_parts. Если она включена (ttl_only_drop_parts=1), ClickHouse удаляет целую часть, только когда все строки в ней просрочены по TTL. Если хотя бы одна строка в части ещё не просрочена, часть останется на диске, и место не освободится. Если же настройку оставить выключенной, ClickHouse выполняет мутацию: для удаления даже одной строки требуется переписать всю упомянутую часть. Это приводит к увеличению потребления ресурсов, поскольку каждый раз создаётся новая часть с оставшимися данными, а старая заменяется новой. Поэтому при выборе TTL и схемы партиционирования важно, чтобы части содержали данные только за нужный интервал (например, день или неделю), иначе удаление будет происходить не так быстро, как ожидается. Подробнее о поведении TTL и настройках см. в документации.

• По умолчанию в ClickHouse при изменении TTL (через ALTER MODIFY TTL) для старых партиций автоматически запускается процесс изменения данных, что на больших таблицах может привести к значительному уменьшению свободного места на диске. Поэтому при работе с большими таблицами можно отключить эту функцию, установив параметр materialize_ttl_after_modify = 0 — в этом случае автоматическое применение TTL к уже существующим данным происходить не будет.

• По возможности избегайте использования global join вместо этого выполняйте джойн глобальной таблицы с локальной, если у обеих таблиц совпадает ключ шардирования.

Плохо (GLOBAL JOIN):

SELECT ...
FROM distributed_table_1
GLOBAL JOIN distributed_table_2 ON ...

В этом случае правая таблица собирается на одном сервере и рассылается по всем нодам, что неэффективно при больших объёмах данных.

Хорошо (локальный JOIN):

SELECT ...
FROM distributed_table_1
JOIN table_2_local ON ...

Каждая нода работает только со своими данными, что минимизирует сетевой трафик и ускоряет выполнение запроса (подробнее).

Если у таблиц разные ключи шардирования, то без global join не обойтись.

• Изменение параметров ORDER BY и PARTITION BY в ClickHouse — это небыстрый процесс, так как требует создания новой таблицы и копирования в неё данных. По возможности стоит выделить дополнительное время для тщательного обдумывания этих параметров.

Заключение

В итоге миграция на новый шардированный кластер ClickHouse оказалась профессиональным вызовом для всей команды. Изменения в архитектуре, новая схема обновления данных и разработка запросов к таблицам с движком Distributed потребовали некоторого времени, за которое наша команда смогла приспособиться к работе на новом шардированном кластере.

Если говорить кратко, то шардирование базы выгодно, когда:

• объём данных превышает возможности одного узла;

• нагрузка на чтение или запись слишком велика для одного узла или реплик, что ведёт к задержкам ответа;

• пропускная способность сети одного узла недостаточна, что замедляет отклик.

Однако, по моему мнению, прежде чем решать вопрос о том, нужен ли вам шардинг, лучше испробовать другие варианты масштабирования баз данных, учитывая, что шардинг — это подход, построение которого требует значительных ресурсов и времени команды. Работа с шардированным ClickHouse действительно значительно сложнее, чем с обычным, нешардированным. Написание запросов становится гораздо более трудоёмким, серьёзно возрастает риск ошибок — уровень сложности существенно выше. 

Если можно обойтись без шардирования, не создавайте себе лишних трудностей. Мы пошли по пути постепенного роста: сначала выстроили сильную экспертизу по ClickHouse без шардирования и только потом начали внедрять шардирование. Нам попадались примеры команд, которые сразу пытались работать с шардированным ClickHouse, но из-за высокой сложности их проекты попросту останавливались, и командам приходилось возвращаться к нешардированному варианту.

На текущий момент на шардированном кластере мы имеем следующие цифры:

• 3.2 ТБ сжатых данных;

• 450 тысяч вызовов из семейств дашбордов;

• 70 активных Airflow DAG'ов с обновлением от 1 до 15 минут;

• источники:

  • 70 топиков;

  • 25 gRPC-ручек;

• более 950 объектов в базе данных; 

• 40 дашбордов.

В целом миграция на новый кластер позволила повысить эффективность обработки данных и обеспечить масштабируемость системы, что важно для поддержки растущих объёмов данных и требований к производительности. 

Дисклеймер. Статья является кратким изложением нашего опыта по переходу на шардирование. Мы сосредоточились на ключевых моментах, детали которых планируем раскрыть в следующих статьях.