Нередким случаем, а тем более для данных, которые хранятся в аналитических базах, обычно это сырые данные, является то, что они перестают помещаться в один сервер. Также для обеспечения доступности данные должны реплицироваться на другие сервера, делается это на случай выхода из строя одного (или нескольких) серверов, а также для возможности масштабирования чтения отправляя запросы на разные сервера. ClickHouse предоставляет возможности “шардирования” и репликации данных.
В отличие от большинства СУБД репликация в ClickHouse не основана на журнале транзакций. Данные в ClickHouse пишутся блоками, после получения блока данных ClickHouse “регистрирует” его в Zookeeper, сервера видят недостающий блок и синхронизируются между собой, более подробно о репликации можно прочесть в официальной документации. Соответственно чтоб репликация заработала вам необходим Zookeeper не старее 3.4.5., а также необходимо создать реплицируемые таблицы, репликация в ClickHouse работает на уровне таблиц и является двунаправленной, не важно в какой из серверов вы производите запись.
Разберем официальный пример:
ReplicatedMergeTree('/clickhouse/tables/{layer}-{shard}/hits', '{replica}', EventDate, intHash32(UserID), (CounterID, EventDate, intHash32(UserID), EventTime), 8192)
Подстановки в фигурных скобках служат для подстановки переменных из секции macros.
<macros>
<layer>05</layer>
<shard>02</shard>
<replica>example05-02-1.yandex.ru</replica>
</macros>Теперь давайте попробуем разобраться со всем этим.
Первым параметром движка ReplicatedMergeTree является идентификатор таблицы, если вы используете шардирование, о нём поговорим ниже, то вам нужно указать уникальное имя таблицы в рамках шарда, для этого удобно использовать макрос {shard}. Вторым параметром {replica} идёт идентификатор сервера, этот идентификатор нужен, чтоб понять какие данные и на каком сервере есть, этот идентификатор должен быть уникальным. Чтоб не путаться можно заменить его на server_id или на любой другой удобный для вас.
Важно. ClickHouse не реплицирует запросы на создание таблиц их необходимо выполнить самостоятельно на всех серверах.
По умолчанию репликация в ClickHouse является асинхронной, данные записываются на один сервер, после чего, когда именно не гарантируется, они попадут на другие. При этом возможна ситуация, что данные были записаны, после чего сервер вышел из строя и потерял данные, в этом случае восстановить данные не получится. Существует настройка меняющая данное поведение — insert_quorum, в ней указывается число реплик в которые должны быть записаны данные после которого сервер ответит что запись была произведена успешно.
В случае если вы хотите быть уверенными в том, что читаете актуальные данные, вы можете выставить настройку select_sequential_consistency в 1. При данной настройке выполняемый запрос выбросит исключение если на реплике ещё нет данных которые были записаны с insert_quorum.
Репликация позволяет создать копии данных, если необходимо хранить объемы данных которые превышают объем одного сервера ClickHouse предоставляет для этого Distributed таблицы и шардирование.
При шардировании каждый сервер хранит свой набор данных (естественно он может иметь реплики, но для простоты мы это сейчас опустим). Данные между шардами распределяются либо по какому-то ключу, напирем по идентификатору пользователя, или равномерно. Для того, чтоб прочитать данные со всех шардов и сформировать единый результат в ClickHouse существуют Distributed таблицы.
Такие таблицы не хранят данные, они лишь отправляют запросы на чтение на все сервера кластера и объединяют/досчитывают их результат, поэтому обращение к разным шардам происходит прозрачно.
Пример конфигурации кластера:
<?xml version="1.0"?>
<yandex>
<remote_servers> <!-- Имя кластера. Указывается при создании Distributed таблицы и распределенных DDL-->
<cluster>
<shard> <!-- Это шард-->
<replica>
<host>replica_1_shard_1</host>
<port>9000</port>
</replica>
<replica>
<host>replica_2_shard_1</host>
<port>9000</port>
</replica>
</shard>
<shard> <!-- Это шард-->
<replica>
<host>replica_1_shard_2</host>
<port>9000</port>
</replica>
<replica>
<host>replica_2_shard_2</host>
<port>9000</port>
</replica>
</shard>
</cluster>
</remote_servers>
</yandex>При заданой конфигурации кластера будут работать не только Distributed таблицы, но и распределенные DDL-запросы.
Частой ошибкой является то, что часто реплики одного шарда вписывают в другой, либо дублируют одну и ту же реплику несколько раз.
Ещё одна возможность Distributed таблиц — это запись в них. Вы можете писать в Distributed таблицу, при этом ClickHouse сам запишет данные в нужные шарды в зависимости от выбранного вами при создании таблицы ключа шардирования.
Distributed(cluster_name, database, table[, sharding_key])
Важно. По умолчанию если вы пишите в Distributed таблицу сервер ClickHouse будет писать во все сервера шарда, если вы пишите в реплицируемую таблицу данное поведение стоит отключить указав в настройках
internal_replicationравнуюtrue. Также вы можете указать параметрinsert_distributed_syncравную1, в этом случае вы получите ответ только когда данные будут отправлены на все узлы кластера.
<shard> <!-- Это шард-->
<internal_replication>true</internal_replication><!-- пишем только в одну из реплик шарда -->
<replica>
<host>replica_1_shard_2</host>
<port>9000</port>
</replica>
<replica>
<host>replica_2_shard_2</host>
<port>9000</port>
</replica>
</shard>Так как в ClickHouse все сервера равны, Distributed таблицы создаются на всех серверах, и читать или писать вы можете в любой из них.
Однако, если вы можете самостоятельно организовать запись в нужный шард — это будет лучше, чем писать в Distributed таблицы. Во первых вы сможете при необходимости легко останавливать запись в шард на время обслуживания, либо для других целей. Во вторых запись напрямую гораздо производительнее чем запись в Distributed таблицу.