Масштабирование кластера: шарды, реплики, ребалансировка

Масштабирование ClickHouse кластера требует понимания ключевого ограничения: добавление нового шарда не приводит к автоматическому перераспределению исторических данных. Новые INSERT-ы пойдут на новый шард, но существующие данные остаются на старых шардах. Планирование ручного rebalancing — неотъемлемая часть операций по масштабированию.

Добавление нового шарда

При добавлении нового шарда нужно обновить конфигурацию кластера на всех узлах:

<!-- /etc/clickhouse-server/config.d/remote_servers.xml (обновлённая конфигурация) -->
<clickhouse>
    <remote_servers>
        <mycluster>
            <!-- Существующий шард 1 -->
            <shard>
                <internal_replication>true</internal_replication>
                <replica><host>clickhouse-01</host><port>9000</port></replica>
                <replica><host>clickhouse-02</host><port>9000</port></replica>
            </shard>
            <!-- Новый шард 2 -->
            <shard>
                <internal_replication>true</internal_replication>
                <replica><host>clickhouse-03</host><port>9000</port></replica>
                <replica><host>clickhouse-04</host><port>9000</port></replica>
            </shard>
        </mycluster>
    </remote_servers>
</clickhouse>

-- Создать ReplicatedMergeTree таблицу на новых узлах
CREATE TABLE events ON CLUSTER 'mycluster' (
    event_time DateTime,
    user_id    UInt64,
    action     LowCardinality(String)
) ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/events', '{replica}')
PARTITION BY toYYYYMM(event_time)
ORDER BY (user_id, event_time);

-- После этого новые INSERT через Distributed таблицу пойдут на все шарды
-- Но исторические данные (до добавления нового шарда) остаются на старых шардах

Диаграмма: процесс масштабирования кластера

Методы rebalancing исторических данных

Метод 1: DETACH PARTITION + ATTACH

Наиболее быстрый подход для перемещения конкретных партиций:

-- Шаг 1: На исходном шарде -- отсоединить партицию
-- (данные остаются на диске, но недоступны для запросов)
ALTER TABLE events ON CLUSTER 'shard1' DETACH PARTITION '202401';

-- Шаг 2: Скопировать файлы партиции на новый шард
-- (через rsync, scp или аналоги)
-- Директория: /var/lib/clickhouse/data/default/events/detached/

-- Шаг 3: На новом шарде -- подключить партицию
ALTER TABLE events ON CLUSTER 'shard2' ATTACH PARTITION '202401';

-- Шаг 4: Проверить, что данные доступны
SELECT count(), min(event_time), max(event_time)
FROM events
WHERE toYYYYMM(event_time) = 202401;

Метод 2: INSERT INTO SELECT через Distributed (resharding)

Для перераспределения данных с применением новой sharding key:

-- Создать новую Distributed таблицу, указывающую на расширенный кластер
CREATE TABLE events_dist_new ON CLUSTER 'mycluster_extended' AS events
ENGINE = Distributed('mycluster_extended', 'default', 'events', user_id);

-- Перелить данные с новым распределением (resharding)
-- ВНИМАНИЕ: высокая нагрузка на сеть и I/O
INSERT INTO events_dist_new
SELECT * FROM events
WHERE toYYYYMM(event_time) BETWEEN 202301 AND 202312;

Метод 3: clickhouse-copier

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

<!-- copier-task.xml -->
<clickhouse>
    <remote_servers>
        <!-- Описание кластеров source и destination -->
    </remote_servers>
    <max_workers>4</max_workers>
    <settings_pull>
        <readonly>1</readonly>
    </settings_pull>
    <tables>
        <table_hits>
            <cluster_pull>source_cluster</cluster_pull>
            <database_pull>default</database_pull>
            <table_pull>events</table_pull>
            <cluster_push>destination_cluster</cluster_push>
            <database_push>default</database_push>
            <table_push>events</table_push>
            <enabled_partitions>
                <partition>'2024-01'</partition>
                <partition>'2024-02'</partition>
            </enabled_partitions>
            <sharding_key>cityHash64(user_id)</sharding_key>
        </table_hits>
    </tables>
</clickhouse>

# Запуск clickhouse-copier
clickhouse-copier \
    --config /etc/clickhouse-copier/copier.xml \
    --task-path /clickhouse/copier-task \
    --base-dir /tmp/copier

Shard weights: временный перекос нагрузки

Shard weights позволяют направлять больше INSERT-ов на конкретный шард — полезно при добавлении нового более мощного шарда:

<!-- remote_servers.xml: временный перекос в пользу нового шарда -->
<shard>
    <weight>1</weight>  <!-- Старый шард: меньший вес -->
    <internal_replication>true</internal_replication>
    <replica><host>clickhouse-01</host><port>9000</port></replica>
</shard>
<shard>
    <weight>3</weight>  <!-- Новый шард: больший вес, получает 75% INSERT-ов -->
    <internal_replication>true</internal_replication>
    <replica><host>clickhouse-03</host><port>9000</port></replica>
</shard>

Топологии кластера (1S_3R, 2S_2R, 3S_2R) рассмотрены подробно в Модуле 09 урок 04.

Добавление реплики: автоматический fetch

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

-- Создать ReplicatedMergeTree таблицу с тем же ZooKeeper path, что и у существующих реплик
-- (на новом узле clickhouse-05, который присоединяется к существующему шарду)
CREATE TABLE events (
    event_time DateTime,
    user_id    UInt64,
    action     LowCardinality(String)
) ENGINE = ReplicatedMergeTree(
    '/clickhouse/tables/shard1/events',  -- Тот же ZK-путь, что у реплик clickhouse-01 и clickhouse-02
    'replica3'                           -- Уникальное имя новой реплики
)
PARTITION BY toYYYYMM(event_time)
ORDER BY (user_id, event_time);

-- ClickHouse автоматически начнёт GET_PART fetches для синхронизации данных
-- Прогресс можно отслеживать через system.replicas
SELECT
    database,
    table,
    replica_name,
    queue_size,
    inserts_in_queue,
    merges_in_queue
FROM system.replicas
WHERE database = 'default' AND table = 'events';

Ключевые выводы

1. Нет авто-rebalancing при добавлении шарда: Исторические данные остаются на старых шардах. Новые INSERT-ы распределяются по всем шардам только после обновления конфигурации.
2. DETACH PARTITION + ATTACH — наиболее быстрый метод перемещения конкретных партиций между шардами без дополнительной нагрузки на CPU (файлы копируются, не переписываются).
3. INSERT INTO SELECT через Distributed создаёт значительную сетевую нагрузку: подходит для небольших объёмов или когда нужно изменить sharding key при переносе.
4. clickhouse-copier — инструмент для масштабного переноса данных с поддержкой параллелизма, возобновления и мониторинга через ZooKeeper.
5. Добавление реплики к существующему шарду запускает автоматический GET_PART fetch через Keeper: данные синхронизируются без ручного вмешательства.