Distributed ClickHouse: репликация, шардирование, план запроса

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

Если ты используешь managed-кластер (ClickHouse Cloud, Yandex Managed CH, Aiven), большую часть из этого делает за тебя облако. Но понимать всё равно полезно: ты быстро упрёшься в стену, если не знаешь, что под капотом. И на собесе на Senior-позицию это спрашивают.

Пререкизит: урок 01 про MergeTree, sparse index, materialized views. Если что-то из этого пока не очевидно, вернись.

0. Когда single-node перестаёт хватать

Сначала проверь, что ты не оптимизируешь до того, как стало больно. Один современный сервер с 32 vCPU, 256 GB RAM и NVMe легко тянет 10-50 ТБ горячих данных и десятки тысяч запросов в секунду. Это много.

Сигналы, что single-node действительно мало:

Disk size. Данные не помещаются на один диск, и cold storage (S3 tiering) не выручает.
Read concurrency. Запросы стоят в очереди, потому что общее CPU кончилось. SELECT count() FROM system.processes стабильно показывает десятки активных.
Ingest throughput. Один писатель упирается в диск или в CPU на компрессию, INSERT не успевают.
HA-требование. Бизнес не готов терять час даунтайма при потере одного диска.

Под каждый сигнал свой инструмент:

Disk size, sharding (горизонтальное разделение данных).
Read concurrency, replication (читать с нескольких реплик параллельно).
Ingest, sharding + клиент-сайд балансировка.
HA, replication.

Чаще всего нужно и то, и другое. Архитектура, которая держится годами в проде, это N шардов по 2-3 реплики каждый.

1. Replication через ClickHouse Keeper

Реплика, это копия таблицы на другом узле, которая получает все INSERT и DDL автоматически. Чтение можно балансировать между репликами, запись приходит на одну, а другие подтягивают.

ClickHouse использует внешний координатор, чтобы реплики договаривались о порядке операций. Раньше это был ZooKeeper (Java, наследие Hadoop). С 2021 года появился собственный ClickHouse Keeper, drop-in замена ZooKeeper, написанная на C++.

1.1 Зачем нужен внешний координатор

ClickHouse реплики между собой консенсус не делают. Они оба смотрят на Keeper, который держит:

список таблиц и их схем,
очередь операций (INSERT блок такой-то, ALTER такой-то),
состояние каждой реплики (что уже применили, что нет).

Когда реплика A получает INSERT, она:

Записывает блок данных к себе локально как новый part.
Регистрирует в Keeper запись “реплика A выложила part ‘all_42_42_0’”.
Реплика B видит запись в Keeper, скачивает part с реплики A через HTTP-интерфейс.
Когда B применила, отмечает в Keeper “applied”.

Это eventually consistent репликация. INSERT возвращает успех клиенту после шага 1 (или после шага 2, если выставлен insert_quorum). Реплика B может отстать на секунды или минуты, если у неё нагрузка.

1.2 Почему Keeper, а не ZooKeeper

ZooKeeper достался с эпохи Hadoop, когда “JVM-стек был стандартом”. В контексте ClickHouse это:

Дополнительная JVM, которую надо тюнить отдельно (heap, GC).
Отдельные навыки эксплуатации (zk-операторы для k8s).
Известные проблемы со скоростью на больших нагрузках.

ClickHouse Keeper использует Raft вместо ZAB (что использует ZooKeeper), компактен, написан на C++. Drop-in совместим по протоколу клиента (для существующих реплицированных таблиц переезд прозрачный).

Ещё одна важная деталь: для production кластера нужно минимум 3 узла Keeper (для кворума 2/3). 1 узел не отказоустойчив, 2 не дают кворума при падении одного. Чётное количество узлов в Raft, антипаттерн.

1.3 ReplicatedMergeTree

Чтобы таблица была реплицирована, в CREATE TABLE указывается одно из семейств Replicated*MergeTree:

CREATE TABLE events_local ON CLUSTER my_cluster (
  ts DateTime, region LowCardinality(String), revenue Float64
)
ENGINE = ReplicatedMergeTree(
  '/clickhouse/tables/{shard}/events_local',  -- путь в Keeper
  '{replica}'                                 -- идентификатор реплики
)
PARTITION BY toYYYYMM(ts)
ORDER BY (region, ts);

{shard} и {replica} это макросы, которые ClickHouse подставляет из конфига (/etc/clickhouse-server/config.d/macros.xml):

<clickhouse>
  <macros>
    <shard>shard_1</shard>
    <replica>replica_1</replica>
  </macros>
</clickhouse>

Каждая реплика подставляет свои значения. Путь /clickhouse/tables/{shard}/events_local это общий znode в Keeper для одного шарда: реплики одного шарда смотрят на один znode и через него договариваются.

ON CLUSTER my_cluster это distributed DDL: команда уйдёт на все узлы кластера, не надо вручную.

1.4 insert_quorum

Дефолт ReplicatedMergeTree, eventually consistent: INSERT возвращает успех после записи на одну реплику. Если эта реплика умрёт до того, как успеет реплицировать, данные потеряются.

Для важных данных можно требовать insert_quorum:

SET insert_quorum = 2;          -- INSERT успешен только после записи на 2 реплики
SET insert_quorum_timeout = 60; -- ждать максимум 60 сек
INSERT INTO events_local VALUES ...;

Цена, latency. INSERT теперь блокируется, пока не получит подтверждения от двух реплик. Для high-throughput pipelines это убивает производительность; для биллинговых событий, скажем, обязательно.

1.5 Чтение с реплик

При запросе через Distributed engine (см. дальше) ClickHouse сам балансирует чтение между репликами в шарде. Стратегии настраиваются:

random (дефолт), случайная реплика.
nearest_hostname, ближайшая по топологии (по совпадению префикса hostname).
in_order, по порядку из конфига.
first_or_random, первая в списке, если жива.

Также есть parallel_replicas, разбивает запрос между репликами одного шарда (не только между шардами). Полезна, когда у тебя мало шардов, но много реплик; обычно достаточно использовать стандартный режим.

2. Sharding и Distributed engine

Шардирование, это разделение данных между узлами. Каждый узел держит только свою часть, и для запроса к полной таблице кто-то должен отправить запрос всем шардам и сложить результаты.

2.1 Логическая схема

Client
  │
  ▼
Distributed table (фасад, без данных)
  │
  ├──> shard_1 (replica_1, replica_2)  -- хранит часть данных
  ├──> shard_2 (replica_1, replica_2)
  └──> shard_3 (replica_1, replica_2)

Данные физически лежат в локальных ReplicatedMergeTree таблицах на каждом узле (по одной копии на реплику). Distributed table это просто прокси-таблица без своих данных, она знает топологию кластера и умеет fan-out на шарды.

2.2 Конфиг кластера

В /etc/clickhouse-server/config.d/cluster.xml:

<clickhouse>
  <remote_servers>
    <my_cluster>
      <shard>
        <internal_replication>true</internal_replication>
        <replica>
          <host>ch-shard1-replica1</host>
          <port>9000</port>
        </replica>
        <replica>
          <host>ch-shard1-replica2</host>
          <port>9000</port>
        </replica>
      </shard>
      <shard>
        <internal_replication>true</internal_replication>
        <replica>
          <host>ch-shard2-replica1</host>
          <port>9000</port>
        </replica>
        <replica>
          <host>ch-shard2-replica2</host>
          <port>9000</port>
        </replica>
      </shard>
    </my_cluster>
  </remote_servers>
</clickhouse>

internal_replication = true важно: при INSERT в Distributed table данные пишутся только на одну реплику шарда, и дальше Replication через Keeper копирует на остальные. Если выставить false, Distributed сам пишет во все реплики, что приводит к двойной записи и проблемам с дедупликацией.

2.3 Distributed table

CREATE TABLE events ON CLUSTER my_cluster AS events_local
ENGINE = Distributed(
  my_cluster,           -- имя кластера из конфига
  default,              -- database
  events_local,         -- локальная таблица
  cityHash64(user_id)   -- sharding key
);

cityHash64(user_id) это sharding key: при INSERT через Distributed table значение хеша по модулю числа шардов определяет шард-получатель.

Хорошие sharding key:

Hash от user_id или session_id, равномерное распределение, плюс данные одного юзера ложатся на один шард (полезно для JOIN).
rand(), идеально равномерное, но теряется локальность.

Плохие sharding key:

toYYYYMM(ts), все свежие данные на один шард, который захлёбывается.
region, скос, потому что регионы неравномерны.

Sharding key выбирается один раз и навсегда. Менять его, это ребалансировка всех данных, что почти невозможно сделать без даунтайма. Подумай дважды.

2.4 Альтернатива: писать локально, читать через Distributed

Распространённый паттерн: клиенты пишут прямо в локальные таблицы (с балансировкой шарда на стороне приложения), а Distributed table используется только для чтения.

Плюсы: меньше нагрузки на Distributed coordinator, проще контролировать sharding. Минусы: клиенту надо знать топологию.

Этот паттерн особенно популярен с Kafka: на каждом шарде стоит свой consumer, который читает только свою партицию топика и пишет локально. Sharding по партиции Kafka топика становится sharding по шардам ClickHouse.

3. Distributed query execution

Самое интересное, это что происходит, когда запрос приходит на Distributed table.

3.1 Roles: coordinator и executors

Любой узел может стать coordinator, тот, кто принял запрос. Coordinator:

Парсит SQL.
Применяет partition pruning и оптимизацию (часть до шардов).
Превращает запрос в subquery для каждого шарда: типично “сделай частичную агрегацию по своей части данных”.
Параллельно отправляет subquery на все шарды.
Каждый шард (его одна реплика) выполняет частичную агрегацию.
Coordinator собирает частичные результаты, делает финальный merge, возвращает клиенту.

Это scatter-gather модель. Она хорошо работает для агрегаций (COUNT, SUM, GROUP BY), потому что финальный merge простой. Хуже работает для JOIN и DISTINCT (см. дальше).

3.2 Partial aggregation

Для запроса SELECT region, SUM(revenue) FROM events GROUP BY region:

Coordinator -> shards: SELECT region, SUM(revenue) FROM events_local GROUP BY region
Each shard:           returns partial sums per region
Coordinator:          UNION ALL + GROUP BY region + SUM(partial_sums)

Сетевой трафик: только частичные результаты (одна строка на каждый region из каждого шарда), не сырые события. Если в данных 10 регионов и 5 шардов, это 50 строк по сети. Очень дёшево.

3.3 GLOBAL IN и GLOBAL JOIN

Для подзапросов есть тонкость. Обычный IN:

SELECT count() FROM events WHERE user_id IN (SELECT user_id FROM premium_users)

Coordinator отправит этот запрос на каждый шард, и каждый шард будет выполнять подзапрос локально на своих данных. Если premium_users тоже Distributed, получается N×N запросов (каждый шард опросит каждый шард). Это квадратичный взрыв.

Решение, GLOBAL IN:

SELECT count() FROM events WHERE user_id GLOBAL IN (SELECT user_id FROM premium_users)

Coordinator выполняет подзапрос один раз, получает результат, и отправляет на каждый шард уже готовый список (как литерал). Линейная сложность.

То же самое для JOIN: GLOBAL JOIN вместо обычного JOIN, когда правая часть это distributed подзапрос.

Правило: если в подзапросе или JOIN-righthand есть Distributed таблица, ставь GLOBAL.

урок из курса

Урок из курса «Data Engineering». Курс покупается один раз и остаётся навсегда: 500 ₽.

Оплата пока не подключена: премиум и курсы открываются промокодом.

войти ввести промокод