Инструкция по созданию реплицируемых и distributed таблиц в Clickhouse кластере

В СУБД Clickhouse репликация данных настраивается на уровне таблиц, в отличие от традиционных, где это обычно уровень БД или даже инстанса. То есть для того, чтобы данные реплицировались на другие хосты в кластере, необходимо правильным образом создать таблицу. Архитектура Clickhouse подразумевает, что таблицы с одинаковым именем реплицируются между двумя репликами (то есть имеют одинаковые данные), а между шардами никак не связаны, то есть данные “шардируются”. При этом, к сожалению, это никак не контролируется Clickhouse, поэтому вам надо самостоятельно  всё создать/настроить правильно.

Итак, у нас в наличии кластер Clickhouse, например с 2 шардами — в каждом 2 реплики. Имя кластера Clickhouse в Linx Cloud всегда cluster. Выполним запрос:

CREATE TABLE table_name ON CLUSTER cluster
(
    EventDate DateTime,
    CounterID UInt32,
    UserID UInt32
) ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/table_name', '{replica}')
PARTITION BY toYYYYMM(EventDate)
ORDER BY (CounterID, EventDate, intHash32(UserID))
SAMPLE BY intHash32(UserID);

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

Давайте разберем особенности этой команды. Здесь используется движок таблицы ReplicatedMergeTree. Этот движок и другие из семейства Replicated рассчитаны на то, чтобы публиковать изменения в таблице в ZooKeeper. Соответственно, аргументами и являются параметры для сохранения этой информации в Zookeeper. Первая часть — это путь, вторая название ключа, в который публикуется информация. В фигурных скобках указаны макросы, макросы определяются в конфиге у каждого сервера. В кластере Linx Cloud они уже будут сконфигурированы соответственно составу кластера, так что можно указывать прямо так, как в примере.

В итоге серверы реплики одного шарда будут иметь одинаковый путь в ZK, и будут применять изменения с других серверов и публиковать свои (двусторонняя репликация). Получается что можно обратиться к одному из серверов и получить актуальные данные, но как получить данные по всем шардам? Для удобства в этом случае можно использовать тип таблицы Distributed.

CREATE TABLE distributed ON CLUSTER cluster
(
    EventDate DateTime,
    CounterID UInt32,
    UserID UInt32
) ENGINE = Distributed(cluster, default, table_name, UserID);

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

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

Во-первых, вы можете самостоятельно определять, на какие серверы какие данные записывать, и выполнять запись непосредственно на каждый шард. То есть делать INSERT в те таблицы, на которые «смотрит» распределённая таблица. Это наиболее гибкое решение, поскольку вы можете использовать любую схему шардирования, которая может быть нетривиальной из-за требований предметной области.

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

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