Шардирование БД в Clickhouse

Понятие шардирования

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

Установка Clickhouse

Произвести установку сервера Clickhouse на трех выделенных серверах согласно официальной документации в соответствии с выбранной ОС. На двух нодах размещаются сами шарды ClickHouse и ClickHouse Keeper, на третьей - только ClickHouse Keeper.

Нода	Описание
chnode1	Шард 1 + ClickHouse Keeper
chnode2	Шард 2 + ClickHouse Keeper
chnode3	ClickHouse Keeper

Конфигурация Clickhouse для нод chnode1 и chnode2

Произвести конфигурацию каждого сервера Clickhouse:

ℹ️ Конфигурация сервера Clickhouse выполняется путем добавления конфигурационных файлов в формате .xml в следующие директории:

• /etc/clickhouse-server/config.d/ - конфигурация сервера
• /etc/clickhouse-server/users.d/ - конфигурация пользователей и ролей

Сеть и логгирование

sudo nano /etc/clickhouse-server/config.d/network-and-logging.xml

<clickhouse>
        <logger>
                <level>information</level>
                <log>/var/log/clickhouse-server/clickhouse-server.log</log>
                <errorlog>/var/log/clickhouse-server/clickhouse-server.err.log</errorlog>
                <size>1000M</size>
                <count>3</count>
        </logger>
        <display_name>clickhouse</display_name>
        <listen_host>0.0.0.0</listen_host>
        <http_port>8123</http_port>
        <tcp_port>9000</tcp_port>
        <interserver_http_port>9009</interserver_http_port>
</clickhouse>

ClickHouse Keeper

Clickhouse Keeper используется для координации распределенных запросов. Вместо Clickhouse Keeper может использоваться Apache Zookeeper. Конфигурация ниже активирует Clickhouse Keeper, который слушает порт 9181. Кластер ClickHouse Keeper должен состоять из нечетного количества нод для обеспечения кворума, не меньше трех. Для большинства инсталляций подойдут три ноды ClickHouse Keeper, большее число нод потребуется в случае особых требований к отказоустойчивости кластера ClickHouse Keeper. Подробнее см. в официальной документации.

sudo nano /etc/clickhouse-server/config.d/enable-keeper.xml

<clickhouse>
  <keeper_server>
    <tcp_port>9181</tcp_port>
    <!--
        server_id должен быть уникальным для каждого инстанса Clickhouse. В данном примере <server_id>1</server_id>, для остальных серверов будет <server_id>2</server_id>, <server_id>3</server_id>.
      -->
    <server_id>1</server_id>
    <log_storage_path>/var/lib/clickhouse/coordination/log</log_storage_path>
    <snapshot_storage_path>/var/lib/clickhouse/coordination/snapshots</snapshot_storage_path>

    <coordination_settings>
        <operation_timeout_ms>10000</operation_timeout_ms>
        <session_timeout_ms>30000</session_timeout_ms>
        <raft_logs_level>trace</raft_logs_level>
    </coordination_settings>
    <!--
        RAFT-конфигурация. Одинаковая на всех серверах Clickhouse.
      -->
    <raft_configuration>
        <server>
            <id>1</id>
            <hostname>chnode1</hostname>
            <port>9234</port>
        </server>
        <server>
            <id>2</id>
            <hostname>chnode2</hostname>
            <port>9234</port>
        </server>
        <server>
            <id>3</id>
            <hostname>chnode3</hostname>
            <port>9234</port>
        </server>
    </raft_configuration>
  </keeper_server>
</clickhouse>

Произвести настройку пользователей в Clickhouse. Для этого создать конфигурационный файл со следующим содержимым, указав в теге пароль для администратора 'pult':

sudo nano /etc/clickhouse-server/users.d/pult.xml

<clickhouse>
    <users>
        <default>     
            <access_management>1</access_management>
            <named_collection_control>1</named_collection_control>
            <show_named_collections>1</show_named_collections>
            <show_named_collections_secrets>1</show_named_collections_secrets>
        </default>
        <!-- Создание нового пользователя pult в качестве администратора БД -->
        <pult>
            <password>changeme</password>
            <grants>
                <query>GRANT admin</query>
            </grants>
        </pult>
    </users>
    <!-- Создание роли admin -->
    <roles>
    <admin>
        <grants>
            <query>GRANT ALL ON pult.* WITH GRANT OPTION;</query>
        </grants>
    </admin>
    </roles>
    <!-- Настройка стандартного профиля default для корректного форматирования данных в JSON (необходимо для интеграции с Grafana) -->
    <profiles>
        <default>
            <input_format_json_try_infer_numbers_from_strings>1</input_format_json_try_infer_numbers_from_strings>
            <output_format_json_quote_64bit_floats>1</output_format_json_quote_64bit_floats>
        </default>
    </profiles>
</clickhouse>

Макросы

Макросы упрощают использование запросов DDL. В данной конфигурации определены номер шарда, название кластера и реплики. Название реплики везде будет replica_1, так как шарды имеют только одну реплику. Данную конфигурацию нужно добавить на ноды chnode1 и chnode2 с указание актуального номера шарда.

sudo nano /etc/clickhouse-server/config.d/macros.xml

<clickhouse>
  <macros>
    <shard>1</shard>
    <replica>replica_1</replica>
    <cluster>pult_cluster_2S_1R</cluster>
  </macros>
</clickhouse>

Конфигурация шардирования

Одинаковая на chnode1 и chnode2

sudo nano /etc/clickhouse-server/config.d/remote-servers.xml

<clickhouse>
  <remote_servers replace="true">
    <pult_cluster_2S_1R>
    <secret>mysecretphrase</secret>
        <shard>
            <internal_replication>true</internal_replication>
            <replica>
                <host>chnode1</host>
                <port>9000</port>
            </replica>
        </shard>
            <shard>
            <internal_replication>true</internal_replication>
            <replica>
                <host>chnode2</host>
                <port>9000</port>
            </replica>
        </shard>
    </pult_cluster_2S_1R>
  </remote_servers>
</clickhouse>

Конфигурация соединения с ClickhouseKeeper

Одинаковая на chnode1 и chnode2

sudo nano /etc/clickhouse-server/config.d/remote-servers.xml

<clickhouse>
    <zookeeper>
        <node index="1">
            <host>chnode1</host>
            <port>9181</port>
        </node>
        <node index="2">
            <host>chnode2</host>
            <port>9181</port>
        </node>
        <node index="3">
            <host>chnode3</host>
            <port>9181</port>
        </node>
    </zookeeper>
</clickhouse>

Конфигурация Clickhouse для ноды chnode3

Так как нода chnode3 не используется для хранения данных, данный сервер имеет всего два конфигурационных файла:

Сеть и логгирование

sudo nano /etc/clickhouse-server/config.d/network-and-logging.xml

<clickhouse>
        <logger>
                <level>debug</level>
                <log>/var/log/clickhouse-server/clickhouse-server.log</log>
                <errorlog>/var/log/clickhouse-server/clickhouse-server.err.log</errorlog>
                <size>1000M</size>
                <count>3</count>
        </logger>
        <display_name>clickhouse</display_name>
        <listen_host>0.0.0.0</listen_host>
        <http_port>8123</http_port>
        <tcp_port>9000</tcp_port>
        <interserver_http_port>9009</interserver_http_port>
</clickhouse>

Конфигурация ClickHouse Keeper

sudo nano /etc/clickhouse-server/config.d/enable-keeper.xml 

<clickhouse>
  <keeper_server>
    <tcp_port>9181</tcp_port>
    <server_id>3</server_id>
    <log_storage_path>/var/lib/clickhouse/coordination/log</log_storage_path>
    <snapshot_storage_path>/var/lib/clickhouse/coordination/snapshots</snapshot_storage_path>

    <coordination_settings>
        <operation_timeout_ms>10000</operation_timeout_ms>
        <session_timeout_ms>30000</session_timeout_ms>
        <raft_logs_level>trace</raft_logs_level>
    </coordination_settings>

    <raft_configuration>
        <server>
            <id>1</id>
            <hostname>chnode1</hostname>
            <port>9234</port>
        </server>
        <server>
            <id>2</id>
            <hostname>chnode2</hostname>
            <port>9234</port>
        </server>
        <server>
            <id>3</id>
            <hostname>chnode3</hostname>
            <port>9234</port>
        </server>
    </raft_configuration>
  </keeper_server>
</clickhouse>

Перезапустить службы Clickhouse

sudo systemctl restart clickhouse-server

Создание БД pult в Clickhouse

#Определить переменную окружения с названием кластера
export cluster=pult_cluster_2S_1R;

Выполнить DDL-запрос

clickhouse-client --user pult --password changeme --multiquery "CREATE DATABASE IF NOT EXISTS pult ON CLUSTER {cluster}; \
CREATE TABLE pult.history_shard ON CLUSTER '{cluster}' (`itemid` UInt64, `clock` UInt64 CODEC(Delta(8), ZSTD(1)), `ns` UInt64, `value` Float64) ENGINE = MergeTree PARTITION BY toYYYYMM(CAST(clock, \'date\')) ORDER BY (itemid, clock) TTL CAST(clock, \'date\') + toIntervalMonth(6) SETTINGS index_granularity = 8192; \
CREATE TABLE pult.history_uint_shard ON CLUSTER '{cluster}' (`itemid` UInt64, `clock` UInt64 CODEC(Delta(1), LZ4), `ns` UInt64, `value` UInt64) ENGINE = MergeTree PARTITION BY toYYYYMM(CAST(clock, \'date\')) ORDER BY (itemid, clock) TTL CAST(clock, \'date\') + toIntervalMonth(6) SETTINGS index_granularity = 8192; \
CREATE TABLE pult.history_str_shard ON CLUSTER '{cluster}' (`itemid` UInt64, `clock` UInt64, `ns` UInt64, `value` String) ENGINE = MergeTree PARTITION BY toYYYYMM(CAST(clock, \'date\')) ORDER BY (itemid, clock) TTL CAST(clock, \'date\') + toIntervalMonth(6) SETTINGS index_granularity = 8192; \
CREATE TABLE pult.history_text_shard ON CLUSTER '{cluster}' (`itemid` UInt64, `clock` UInt64, `ns` UInt64, `value` String) ENGINE = MergeTree PARTITION BY toYYYYMM(CAST(clock, \'date\')) ORDER BY (itemid, clock) TTL CAST(clock, \'date\') + toIntervalMonth(6) SETTINGS index_granularity = 8192; \
CREATE MATERIALIZED VIEW pult.trends_shard ON CLUSTER '{cluster}' (`itemid` UInt64, `clock` DateTime(\'Europe/Moscow\'), `num` AggregateFunction(count), `value_min` AggregateFunction(min, Float64), `value_avg` AggregateFunction(avg, Float64), `value_max` AggregateFunction(max, Float64)) ENGINE = AggregatingMergeTree PARTITION BY toYYYYMM(clock) ORDER BY (itemid, toStartOfHour(toDateTime(clock), \'Europe/Moscow\')) SETTINGS index_granularity = 8192 AS SELECT itemid, toStartOfHour(toDateTime(clock), \'Europe/Moscow\') AS clock, countState(*) AS num, minState(value) AS value_min, avgState(value) AS value_avg, maxState(value) AS value_max FROM pult.history GROUP BY itemid, clock; \
CREATE MATERIALIZED VIEW pult.trends_uint_shard ON CLUSTER '{cluster}' (`itemid` UInt64, `clock` DateTime(\'Europe/Moscow\'), `num` AggregateFunction(count), `value_min` AggregateFunction(min, UInt64), `value_avg` AggregateFunction(avg, UInt64), `value_max` AggregateFunction(max, UInt64)) ENGINE = AggregatingMergeTree PARTITION BY toYYYYMM(clock) ORDER BY (itemid, toStartOfInterval(toDateTime(clock, \'Europe/Moscow\'), toIntervalHour(1))) TTL toStartOfInterval(toDateTime(clock, \'Europe/Moscow\'), toIntervalHour(1)) + toIntervalMonth(12) SETTINGS index_granularity = 8192 AS SELECT itemid, toStartOfInterval(toDateTime(clock, \'Europe/Moscow\'), toIntervalHour(1)) AS clock, countState(*) AS num, minState(value) AS value_min, avgState(value) AS value_avg, maxState(value) AS value_max FROM pult.history_uint_shard GROUP BY itemid, clock; \
CREATE TABLE pult.history ON CLUSTER '{cluster}' (`itemid` UInt64, `clock` UInt64, `ns` UInt64, `value` Float64) ENGINE = Distributed(\'{cluster}\', \'pult\', \'history_shard\', rand()); \
CREATE TABLE pult.history_uint ON CLUSTER '{cluster}' (`itemid` UInt64, `clock` UInt64, `ns` UInt64, `value` UInt64) ENGINE = Distributed(\'{cluster}\', \'pult\', \'history_uint_shard\', rand()); \
CREATE TABLE pult.history_str ON CLUSTER '{cluster}' (`itemid` UInt64, `clock` UInt64, `ns` UInt64, `value` String) ENGINE = Distributed(\'{cluster}\', \'pult\', \'history_str_shard\', rand()); \
CREATE TABLE pult.history_log ON CLUSTER '{cluster}' (`itemid` UInt64, `clock` UInt64, `ns` UInt64, `timestamp` UInt64, `logeventid` UInt64, `source` String, `severity` UInt8, `value` String) ENGINE = Distributed(\'{cluster}\', \'pult\', \'history_log_shard\', rand()); \
CREATE TABLE pult.history_text ON CLUSTER '{cluster}' (`itemid` UInt64, `clock` UInt64, `ns` UInt64, `value` String) ENGINE = Distributed(\'{cluster}\', \'pult\', \'history_text_shard\', rand()); \
CREATE TABLE pult.history_bin ON CLUSTER '{cluster}' (`itemid` UInt64, `clock` UInt64 CODEC(Delta, LZ4), `ns` UInt64 CODEC(T64, LZ4), `value` String) ENGINE = Distributed(\'{cluster}\', \'pult\', \'history_bin_shard\', rand());"