ClickHouse — одна из самых популярных систем для анализа данных. По информации TheirStack, этот инструмент использует более 3 700 компаний по всему миру. 

У ClickHouse быстрая аналитика, эффективное сжатие и отличное масштабирование. Но у системы есть и недостатки — ограниченная поддержка UPDATE и DELETE, а также сложная миграция.

Привет, Хабр! Меня зовут Михаил Филимонов, я руковожу разработкой хранилища данных в группе Магнит OMNI. В этой статье я расскажу о проблемах работы с ClickHouse, как их решать и какие инструменты для этого потребуются. 

Архитектура хранилища данных OMNI

Начнём с того, что Magnit OMNI — это группа, которая объединяет цифровые направления «Магнита»:

• Mагнит Плюс. Все бонусы, предложения о работе и другие элементы системы лояльности «Магнита».

• Магнит FoodTech. Всё, что связано с доставкой еды на дом. 

• AdTech. Монетизация рекламы;

• Магнит Маркет. Маркетплейс с низкими ценами и быстрой доставкой.на основе всей сети «Большой Магнит».

Также есть OMNIApp — приложение, через которое можно пользоваться всеми сервисами в едином интерфейсе. 

А теперь подробнее разберём архитектуру хранилища Магнит OMNI. В неё входит:

• Greenplum. Ядро хранилища. Здесь мы строим Data Vault, рассчитываем метрики в режиме «сегодня за вчера» и допускаем сюда аналитиков NetHawk.

• ClickHouse. Используем для аналитики в реальном времени, BI-нагрузки и анализа журналов мобильного приложения. 

• Airflow + DBT. Всё, что касается шаблонизации и фреймворков. Например, мы не раскладываем Data Vault вручную и не пишем витрины самостоятельно — всё это делает за нас фреймворк.

• Kafka + Debezium. Основной метод интеграции данных в хранилище. 

• DataLens. Наш главный BI-инструмент. 

• DBT + OpenMetadata. Базовая связка для Data Catalog — это всё, что касается вопросов Data Governance.

Как мы работаем с ClickHouse

Для начала расскажу о главных проблемах, с которыми мы столкнулись при внедрении ClickHouse. Иначе говоря, разберу несколько «вредных» советов.

Вредный совет №1: не используйте реплики 

Это самая критичная ошибка в промышленной эксплуатации ClickHouse. Вот почему мы её допустили:

• Высокая стоимость. При выборе фактора репликации цена инфраструктуры повышается линейно. Например, коэффициент ×2 удваивает расходы, коэффициент ×3 — утраивает, и так далее.

• Контроль синхронизации реплик. ClickHouse работает асинхронно, поэтому при несогласованности данных один и тот же запрос, отправленный на разные реплики, может вернуть различные результаты. 

• Сложность в адаптации существующей архитектуры. Стандартный MergeTree не поддерживает репликацию, поэтому все таблицы нужно переносить на replicated-движки. Кроме того, в коде приходится менять настройки — например, обычные кластеры на cluster all-replicas.

К чему это привело:

• Отказ одного узла делал недоступным весь кластер.

• Недоступность кластера обрывала все потоки real-time ingestion, и приходилось экстренно загружать данные из сервисных баз с помощью бэкфилов.

В итоге доступность production-кластера составила 89% — это крайне низкий показатель.

Вредный совет №2: используйте мутации

Мутация в ClickHouse — это операции UPDATE и DELETE, которые крайне тяжело обрабатываются системой. Даже в официальной документации прямо сказано: avoid mutations (избегайте мутаций) — но этот совет мы проигнорировали. 

Всё потому, что аналитикам привычнее написать UPDATE или DELETE, чем заменять данные через партиции или использовать специализированные движки, вроде ReplacingMergeTree или AggregatingMergeTree. 

В итоге мы получили следующее:

• каждая модификация выполнялась с пометкой ON CLUSTER и попадала в общую очередь на каждом узле;

• если очередь останавливалась хотя бы на одном хосте, это блокировало работу всего ETL-кластера до завершения проблемной операции;

• создавалась большая очередь распределённых запросов.

Вместо ускоренной аналитики мы получили системные сбои и замедленную обработку данных.

Вредный совет №3: откажитесь от управления схемами

Здесь у нас не было объективных причин, просто организационный недочёт. Мы не создали единых правил и договорённостей при работе со схемами, и вот к чему это привело:

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

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

• передача знаний между аналитиками происходила исключительно в устной форме.

Это плохо отразилось на построении большой production-системы. 

Как мы исправили ситуацию

Мы не стали реанимировать старый ClickHouse, а переехали на новый, построенный правильно с нуля. На это было несколько причин:

• Очень мало свободного места на дисках. Нам не хватало пространства даже на копии таблиц, которые нужно было отрефакторить — например, чтобы поменять движок или топологию.

• Постепенный переход не имел перспектив. Даже если бы мы перевели один домен на новую структуру, где аналитики всё делали правильно — не использовали DELETE/UPDATE и соблюдали паттерны, — то остальные домены специалисты все равно обрабатывали бы по-прежнему. 

• Переход сложно «продать» команде. Аналитиков не интересуют задачи, которые не влияют на их ежедневные результаты — повышение uptime не входит в их зону ответственности. 

Но при наличии второго кластера ситуация меняется. У нас появляется простой административный рычаг: «Не перенесёте таблицы до дедлайна — не получите доступ к данным». Такой подход намного эффективнее.

Что мы сделали при переезде в новый кластер

Мы не просто сменили оборудование, а учли весь предыдущий опыт и закрыли три основные проблемы:

• Внедрили реплики. Теперь нагрузка распределялась между мастером и репликой, а хосты больше не вылетали. Проблема с аптаймом была закрыта.

• Запретили мутации. UPDATE и DELETE для пользователей отключили на уровне прав. После этого очередь распределённых запросов не зависала.

• Навели порядок в схемах. Собрали схемы, оформили песочницы для аналитиков и закрепили ответственных. Это дало прозрачную, задокументированную в Confluence архитектуру. 

Но даже после этого оставались сложности:

• Одновременный отказ хоста и его реплики делал недоступным весь шард.

• Зависание onCluster-запроса (из-за троттлинга дисков) останавливало весь ETL-процесс.

• Изоляция нагрузки оставалась только на уровне настроек, но не оборудования — любая команда могла перегрузить общий диск и повлиять на соседей.

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

Как мы решили проблемы в новом ClickHouse

Если коротко, то мы внедрили шард-группы. Это решение позволило разделить кластер на независимые сегменты, где каждая команда получала свой изолированный контур, со своими ресурсами и данными. 

Это дало нам:

• изоляцию нагрузки;

• прозрачное масштабирование;

• гибкость в управлении ресурсами.

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

Вот как изменилась архитектура нового кластера:

• Подключение пользователей. Раньше пользователь подключался по идентификатору кластера (not realhostname): система сама выбирала узел для выполнения запроса. Теперь система отправляет пользователя прямо в шард-группу его домена — это даёт точность и контроль.

• Разделение нагрузки. Нагрузка стала изолирована на уровне оборудования. У каждого домена есть свой compute и storage, а команды работают только в пределах своих ресурсов и больше не могут влиять на работу друг друга.

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

• onCluster-операции. Раньше onCluster выполнялся на всех узлах кластера. Теперь достаточно указать ID шард-группы, и операция запустится только внутри неё:

truncate table dm.shard_table on cluster '{cluster}';

truncate table dm.shard_table on cluster dm;

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

Кроме того, с помощью таблиц мы решили ещё одну проблему — масштабирование. 

Раньше, чтобы расширить место под команду, приходилось заливать дисками все хосты кластера и вручную перераспределять данные. Теперь расширение выполняется только внутри шард-группы конкретного домена — так можно изолировать ресурсы по доменам и по типу нагрузки. 

А если вертикального масштабирования не хватает, то мы просто добавляем новые шарды в группу домена. Горизонтальное масштабирование при этом всегда остаётся открытым.

Как мы внедрили новую архитектуру

Концепция шард-групп нам понравилась, и мы решили её воплотить. Чтобы ускорить этот процесс, мы сделали следующее:

• Выбрали тестовый домен, чтобы обкатать концепцию и минимизировать риски.

• Пересоздали Distributed-таблицы так, чтобы они «смотрели» только на свою шард-группу и искали физические таблицы в её пределах:

    drop_distr = f"""DROP TABLE {database}.{name} ON CLUSTER c9qt89m7gguffprpua SETTINGS distributed_ddl_task_timeout=-1"""

    print(drop_distr)

    if execute:

        new_ch.execute(drop_distr)

    print(f'distributed table dropped for {database}.{name}')

    recreate_distr = (create_table_query

                      .replace('{cluster}',target_cluster)

                      .replace('c9qt89m7gguffpprrpua',target_cluster)

                      .replace(f'{database}.{name}',f'{database}.{name} on cluster c9qt89m7gguffpprrpua')

                      )

    print(recreate_distr)

    if execute:

        new_ch.execute(recreate_distr)

    print(f'distributed table recreated for {database}.{name}')

• Перенесли данные со всех шардов, исключив из фильтра хосты самого домена, чтобы не дублировать записи.

• Удалили физические таблицы на «чужих» узлах и оставили только те, которые должны быть в своей шард-группе:

    for shard_group in shard_groups_list:

        if shard_group != target_cluster:

            truncate_shard_table = f"""drop table {database}.{name}_shard on cluster {shard_group} SYNC SETTINGS distributed_ddl_task_timeout=-1"""

            print(truncate_shard_table)

            if execute:

                new_ch.execute(truncate_shard_table)

    print(f'table {database}.{name}_shard dropped on other shard groups')

    print('----------------------------------------------------------------------------------------------------------')

А чтобы не «сломать» ETL, мы зашли в репозиторий аналитиков, нашли скрипты и написали bash-команду, которая автоматически подменяла все обращения к кластеру на обращения к ID соответствующей шард-группы.

И вот что мы получили:

• Репликация + изоляция = отказоустойчивость. Если хост выходил из строя вместе с репликой, то последствия ограничивались одной шард-группой. Если «падала» вся группа, то доступ к данным конкретного домена прекращался, но ETL и ad-hoc других доменов работали без перебоев.

• Очередь распределённых запросов. Проблема блокировок перестала быть общей — зависание операции касалось только своей группы. В итоге никакой «каннибализации» нагрузки между доменами.

• Прозрачные инвестиции. Мы начали масштабировать не весь кластер, а только группы конкретных доменов. Это сделало инвестиции понятными и управляемыми. После этого Uptime вырос с 89 % до 99,9 % — и это действительно сильный результат.

Если говорить об объёмах, то RPS у нас невысокий, это не OLTP-нагрузка. Мы не спамим мелкими запросами, а работаем с большими OLAP-задачами и сложными ad-hoc.

Соответственно, и данные измеряются внушительными числами — более 40 ТБ в хранилище и еще 165 ТБ зарезервировано в Яндекс S3 под ClickHouse.

Что мониторить, чтобы не откатиться назад

Очередь распределённых запросов

Мы следим за ней двумя способами — через системную таблицу очереди (ищем незавершенные операции) и через отслеживание зависших мутаций.

Время выполнения запросов

Отслеживаем длительность всех запросов, разделяя их на три типа:

• ETL: определяется по Query ID — ID задачи в Airflow, Dag’a, пользователя и UID;

• BI: фиксируется по HTTP User Agent — например, DataLens;

• Ad-hoc: всё остальное, кроме технических учёток.

Мы можем выбрать диапазон, определить среднее время, 90-й и 99-й перцентили, посмотреть статистику конкретного пользователя и прийти к DWH-команде с точным запросом: «У нас тормозят дашборды, посмотрите».

Заключение

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

Кроме того, ClickHouse сильно ускорил работу с отчётами и витринами. Сейчас 99 % ETL-запросов укладываются в 750 секунд — для наших объёмов это отличный результат.