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

Прежде чем начать, приглашаем вас подписаться на наш блог Хабр, TG-канал DevOps FM, интернет-издание vc.ru и познакомиться с YouTube — мы всегда рады новым друзьям :) Теперь ближе к делу. 

Для начала рассмотрим терминологию. Она базовая, однако в рамках ClickHouse есть определённые тонкости, которые необходимо проговорить заранее.

Реплика

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

Шард

Подмножество данных, в котором шардированные данные разбиваются на непересекающиеся наборы. Допустим, у нас есть таблица с большим количеством логов, которая физически не помещается на одном сервере. Мы можем разбить таблицу на 4 части и разместить на 4 разных серверах. Это позволяет не только уменьшить объём данных на одном инстансе, но и увеличить пропускную способность системы и количество операций ввода/вывода за счёт того, что операции выполняются на всех хостах.

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

Метадата

Информация, которую описывают другие данные и представляют собой структурированные справочники, что помогают сортировать и идентифицировать атрибуты описываемой ими информации.

Отлично, теперь мы можем перейти непосредственно к репликации. Начнём с того, зачем она в целом нужна в ClickHouse. Изначально в распределённых системах репликация решала 3 основные задачи:

• Обеспечение высокой доступности:

Тут всё просто: при потере одного из инстансов второй продолжает работать, и мы можем переключить трафик только на рабочую ноду. Если реплика недоступна, то репликация не падает, и данные на неё запишутся тогда, когда она вернётся в строй. 

• Масштабирование нагрузки;

Тут все так же несложно: при повышении количества запросов или объёма данных мы вводим дополнительный инстанс ClickHouse в систему и масштабируемся горизонтально. С точки зрения топологии, ClickHouse позволяет создать различные архитектуры: вы можете использовать как только шардированые таблицы или только реплицирование таблицы, так и комбинированные вариации, в которых шарды реплицируются.

• Упрощение процесса миграции и обновления;

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

Однако, с ростом проекта репликация ClickHouse превращается из роскоши в необходимость. Почему? Все дело в размерах таблиц: как правило, в ClickHouse хранятся исторические данные, объём которых будет неминуемо расти. С ростом бизнеса объём поступающих данных может спокойно выйти за сотни гигабайт в день. Последнее, что хочется при такой динамике — терять данные при сбое. При малых масштабах данную проблему закрывают бэкапы, однако они покрывают, в лучшем случае, день. То есть при восстановлении из бэкапа потребуется время не только на заливку бэкапа в инстансы,но и на перезапуск всех ETL-процессов, чтобы повторно записать данные, которые были созданы между моментом создания бекапа и до выхода из строя кластера. На практике это занимает значительное количество времени и, как правило, блокирует процессы многих команд. Эту проблему крайне эффективно решает репликация, особенно геораспределённая, но об это чуть позже.

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

Концепция репликации в ClickHouse схожа, однако за хранение этого журнала отвечает не сама база. Для хранения и координации действий репликации в ClickHouse требуется распределённое хранилище данных, которое может гарантировать согласованность состояния. Для этого ClickHouse использует один из двух вариантов: Zookeeper или ClickHouse Keeper.

Keeper не хранит в себе данные таблиц, но содержит метаданные, которые говорят, какие данные должны хранится в каждой реплике. Например, если вы вставляете 10 строк, то ClickHouse создаст парт и ссылка на этот парт будет хранится в Keeper. Все инстансы ClickHouse будут знать, что парт есть в кластере и необходимо найти его в локальной файловой системе. А если его нет локально, то инстанс должен обратиться к другим инстансам, чтобы скопировать его. Корректность метаданных и доступность самого Keeper критически важны для корректной работы репликации: если вы потеряете свой Keeper, то все реплицируемые таблицы уйдут в read-only режим до возвращения Keeper в рабочее состояние. Сейчас внутри ClickHouse есть функция SystemRestoreReplica, которая позволяет восстановить репликация из локальных данных, однако этот вариант стоит использовать только в крайнем случае. Помимо хранения метаданных, ClickHouse также использует Keeper для координации кластера.

В этой статье мы будем рассматривать в основном работу с ZooKeeper, так как это хранилище используется на большей части наших проектов, однако скажем пару слов про ClickHouse Keeper.

ClickHouse Keeper — это замена ZooKeeper, которая поддерживает собственное расширение, оптимизированное под работу с ClickHouse. ClickHouse Keeper можно поддерживать в двух разных установках: в качестве обособленной инсталляции на отдельных серверах или как часть ClickHouse в виде Process Keeper — части бинарного файла ClickHouse clickhouse-server. Несмотря на то, что второй вариант кажется проще и дешевле (так как по сути Keeper уже встроен в ClickHouse), мы настоятельно рекомендуем использовать ClickHouse Keeper как отдельную инсталляцию, так как это повышает отказоустойчивость системы в целом.

Если вы хотите перейти с ZooKeeper на ClickHouse Keeper, стоит помнить 2 нюанса: во-первых, снэпшоты и журналы ClickHouse Keeper имеют несовместимый формат с ZooKeeper, однако можно конвертировать данные Zookeeper в снэпшот ClickHouse Keeper с помощью clickhouse-keeper-converter. Во-вторых, межсерверный протокол ClickHouse Keeper несовместим с ZooKeeper, поэтому создание смешанного кластера ZooKeeper + ClickHouse Keeper невозможно.

Теперь вернемся к Zookeeper и посмотрим на него в рамках кластера ClickHouse. Внутри Zookeeper все данные о кластере будут хранится в директории /clickhouse и в ней будет находятся следующее:

• ddl — все DDL запросы, которые необходимо реплицировать и применять к репликам.

• metadata — хранит часть информации о метаданных реплицируемой таблицы, например первичный ключ, ключ раздела и т. д;

• columns — хранит информацию о полях реплицируемой таблицы;

• block_numbers — хранит все разделы таблицы;

• leader_election — используется для выбора мастера между репликами. При запросе сначала будет выбрана ведущая реплика, а затем на ее основе будет синхронизирована другая реплика. Несколько реплик могут быть лидерами одновременно;

• log — каталог, служащий очередью задач для хранения операций над репликой таблицы;

• blocks — хранит хэш-информацию блоков данных, записанных в таблицу за определенный период времени для дедупликации. Формат следующих дочерних узлов — partition_hash_hash;

• mutation — каталог, служащий очередью задач для хранения операций мутации над реплицированной таблицей;

• replicas/node_hostname — хранит информацию о репликах:

  • is_active — флаг, указывающий, активен ли инстанс или нет, если на сервере произошел сбой, узел не будет существовать, и будет добавлен заново после восстановления;

  • mutation_pointer — хранит следующую задачу в очереди мутаций, которую следует выполнить;

  • log_pointer — хранит следующую задачу в очереди журналов, которую следует выполнить;

  • is_lost — флаг, указывающий, устарела ли реплика, на основании того, обновлён ли указатель log_pointer, при этом 0 — нормально, -1 — устарела, а 1 — находится в процессе восстановления;

  • metadata — хранит информацию о метаданных таблицы, как и metadata выше;

  • columns — хранит информацию о столбцах таблицы;

  • parts — хранит информацию о партах таблицы, каждый блок информации содержит контрольные суммы и информацию о столбцах;

  • queue — временная очередь обработки

Теперь рассмотрим стандартную реализацию кластера ClickHouse. В нашей схеме будет 5 серверов: две машины для мастер-мастер инстансов ClickHouse и три машины для кластера ZooKeeper:

В той схеме используются следующие порты:

• 9000 — стандартный порт работы ClickHouse;

• 9009 — стандартный порт для обмена информацией между реплицируемыми таблицами;

• 2181 — стандартный порт для связи с ZooKeeper;

А теперь возьмем ту же самую схему и сделаем ее геораспределённой:

Наш основной контур ClickHouse находится в одном дата-центре, а резервный — в другом дата-центре и регионе. В резервный контур не идет запись, так как сетевая задержка в таком случае будет крайне заметна. Однако при полном выходе из строя основного контура нам будет достаточно переключить писателей на резервный кластер, который самодостаточен и обладает теми же данными в базах и метаданными в Keeper (не считая тех, что не дошли из-за сети), что и основной кластер.

С основной топологией разобрались, теперь посмотрим на саму реализацию репликации в ClickHouse более подробно.

С точки зрения реализации в ClickHouse репликация — это расширение над семейством движков MergeTree. То есть, когда мы говорим о репликации, подразумевается, что мы получаем обычное поведение конкретной реализации MergeTree, которое может реплицироваться. Проще говоря, мы добавляем Replicated к названию MergeTree движка и получаем его реплицированную версию.

Репликация в ClickHouse работает на основе мастер-мастер таблиц без каких-либо блокировок, как в OLTP базах. Это достигается за счёт архитектуры Shared Nothing, которая распределяет все запросы по отдельным ядрам. Когда вы вставляете что-то на инстансе, это будет асинхронно реплицировано на другие реплики. По умолчанию, запрос INSERT ждёт подтверждения записи только от одной реплики. То есть, если данные были успешно записаны только на одну реплику, и сервер с этой репликой вышел из строя, то записанные данные будут потеряны. Однако для предотвращения подобной ситуации можно включить подтверждение записи от нескольких реплик, используя настройку insert_quorum, но это негативно скажется на скорости работы. Также из-за асинхронности данные на сервере появляются с небольшой задержкой, и она составляет столько времени, сколько требуется для передачи блока сжатых данных по сети, образуя тем самым так называемый лаг репликации.

В рамках MergeTree вы вставляете парты, а ClickHouse будет производит их вставку и слияние с другими партами в фоновом режиме, количество потоков для выполнения фоновых задач можно задать с помощью настройки background_schedule_pool_size. За то, как именно должно производится слияние будет отвечать лидер - одна из таблиц, которую ClickHouse выбирает как ответственную и она отвечает за то, как должно происходить фоновое слияние, используя кластер Keeper и сообщать об этом репликам. Таким образом, в отличие от OLTP баз, в ClickHouse вас не должен волновать статус лидерства реплики. DDL и TRUNCATE операции также координируются, если применены с ON CLUSTER. Таким образом, почти все операции вставки или изменения таблиц будут реплицироваться, кроме следующих:

• CREATE table;

• DROP table;

• DETACH table;

• ATTACH table;

• ALTER TABLE FREEZE;

• ALTER TABLE MOVE TO DISK;

• ALTER TABLE FETCH;

Указанные выше операции являются локальными для инстанса и их необходимо выполнять на всех узлах кластера отдельно.

Если в процессе эксплуатации появляется необходимость перейти с нереплицируемой MergeTree таблицы на её Replicated-версию, то есть два основных варианта:

1) Ручной: создать реплицируемую таблицу с таким же движком и структурой, после чего присоединить партиции через ATTACH одну за одной. Этот вариант работает на всех версиях ClickHouse.  

2) Автоматический: можно добавить флаг репликации в каталог таблицы и ClickHouse автоматически преобразует таблицу в реплицируемую. Эта функция была введена в версии 24.2.

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

Падение репликации

С подобной ситуацией мы сталкивались однажды, однако от этого она не стала более приятной.

В одно дивное зимнее утро система мониторинга уведомила нас о лаге репликации. В логах мелькала ошибка таймаута и выхода за пределы лимита по памяти:

DB::Exception: Received from <Ip>. DB::Exception: Memory limit (total) exceeded: would use 113.01 GiB (attempt to allocate chunk of 5240030 bytes), maximum: 113.01 GiB

Это оказалась миграция, которая имела слишком большой объём операций, вследствие чего после обработки запроса она зависла. Запрос не смог среплицироваться, и произошёл лаг репликации. Это не было большой проблемой: после выполнения миграции репликация бы прошла успешно. А вот что являлось проблемой, так это перезапуск инстанса, на котором выполнялся эта миграция по причине технического сбоя ноды. Это привело к тому, что часть таблиц ушла в Read Only режим. Система пыталась установить соединение с ZooKeeper, который был доступен, но из-за ошибки при взаимодействии с ним было получено исключение. Вероятнее всего это было связано с тем, что тяжелый запрос не были завершён, что не позволило соотнести информацию с ZooKeeper.

Для решения этой проблемы мы завершили эти процессы, после чего вновь перезапустили ClickHouse на втором инстансе. После этого таблицы вышли из Read Only режима, однако появились 3 проблемы в репликации: таблицы содержали больше 300 parts_to_check. 

SELECT

    database,

    table,

    is_leader,

    is_readonly,

    is_session_expired,

    future_parts,

    parts_to_check,

    columns_version,

    queue_size,

    inserts_in_queue,

    merges_in_queue,

    log_max_index,

    log_pointer,

    total_replicas,

    active_replicas

FROM system.replicas

WHERE is_readonly OR is_session_expired OR (future_parts > 20) OR (parts_to_check > 10) OR (queue_size > 20) OR (inserts_in_queue > 10) OR ((log_max_index - log_pointer) > 10) OR (total_replicas < 2) OR (active_replicas < total_replicas)

┌─database─┬─table────────────────┬─is_leader─┬─is_readonly─┬─is_session_expired─┬─future_parts─┬─parts_to_check─┬─columns_version─┬─queue_size─┬─inserts_in_queue─┬─merges_in_queue─┬─log_max_index─┬─log_pointer─┬─total_replicas─┬─active_replicas─┐

│ <db_name>       │ <table_name1> │         0 │           0 │                  0 │            0 │              0 │              -1 │        133 │              129 │               4 │      26974009 │    26974010 │              2 │               2 │

│ <db_name>       │ <table_name2> │         0 │           0 │                  0 │            5 │            333 │              -1 │         10 │                5 │               5 │      73264917 │    73264918 │              2 │               2 │

│ <db_name>       │ <table_name3> │         0 │           0 │                  0 │            0 │              0 │              -1 │        458 │              458 │               0 │       9938773 │     9938774 │              2 │               2 │

└──────────┴─────────────────────────┴───────────┴─────────────┴────────────────────┴──────────────┴────────────────┴─────────────────┴────────────┴────────

──────────┴─────────────────┴─

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

• Если есть незначительные несоответствия, система устранит их путем синхронизации данных с репликами.

• Если прочитать запросом SELECT файлы, которые совпадают по размерам, но где-то посередине имеют изменённые в сравнении с информацией в Keeper байты, то запрос получит исключение о несоответствующей контрольной сумме или размере сжатого блока. Парты добавятся в очередь на проверку (parts_to_check) и при необходимости скачаются с реплик.

• А если локальный набор данных слишком сильно отличается от ожидаемого из Keeper, то сработает механизм безопасности ClickHouse: cервер занесёт это в журнал и откажется запускаться. В нашем случае инстансы ClickHouse не упали, но часть таблиц была повреждена:

<Error> xr.rep_site_recommendation (ReplicatedMergeTreePartCheckThread): No replica has part covering 1664323200_13512_13512_0 and a merge is impossible: we didn't find smaller parts with either the same min block or the same max block.

Начиная с версии ClickHouse 21.7 появился мощный инструмент SYSTEM RESTORE REPLICA, который позволяет восстановить read only реплику, если данные присутствуют, но метаданные Zookeeper потеряны. Но наша версия ClickHouse была старше, поэтому у нас не было возможности воспользоваться им.

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

create /clickhouse/tables/<node_name>/<table_name>/replicas/<replica_name>/flags/force_restore_data

После чего произвели перезапуск ClickHouse. Это решило проблему и вскоре репликация восстановилась.

Подведём итоги:

• Keeper должен находится на отдельных серверах, желательно на SSD дисках 3,5 или 7 узлов;

• Не стоит очищать Keeper вручную;

• Бэкапируйте информацию из Keeper;

• Систематически обновляйте версию ClickHouse для возможности использовать все актуальные инструменты;

• Оптимизируйте свои запросы.

На этом всё. Сегодня мы рассмотрели основные моменты, связанные с репликацией в ClickHouse, а также кратко рассмотрели процесс анализа и восстановления репликации после сбоя. Увидимся в следующей части!