Что такое distributed tables в ClickHouse и как работает шардирование?

Конфигурация кластера

<remote_servers>
  <my_cluster>
    <shard>
      <replica><host>ch-01</host><port>9000</port></replica>
      <replica><host>ch-02</host><port>9000</port></replica>
    </shard>
    <shard>
      <replica><host>ch-03</host><port>9000</port></replica>
      <replica><host>ch-04</host><port>9000</port></replica>
    </shard>
  </my_cluster>
</remote_servers>

Distributed Tables и шардирование в ClickHouse

Distributed — это виртуальный движок таблиц в ClickHouse, который не хранит данные сам, а служит маршрутизатором запросов к локальным таблицам на других шардах кластера. Физические данные живут в таблицах с движком MergeTree (или ReplicatedMergeTree) на каждом шарде.

Архитектура

• Кластер описывается в config.xml или /etc/clickhouse-server/config.d/cluster.xml секцией <remote_servers>.
• Каждый шард содержит одну или несколько реплик.
• Distributed-таблица создаётся поверх локальных таблиц и ссылается на имя кластера.

Создание локальной и Distributed-таблицы

-- На каждом узле: локальная таблица
CREATE TABLE events_local ON CLUSTER my_cluster
(
    event_date Date,
    user_id    UInt64,
    event_type String
)
ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/events', '{replica}')
PARTITION BY toYYYYMM(event_date)
ORDER BY (user_id, event_date);

-- Distributed-обёртка
CREATE TABLE events ON CLUSTER my_cluster
(
    event_date Date,
    user_id    UInt64,
    event_type String
)
ENGINE = Distributed(my_cluster, default, events_local, cityHash64(user_id));

Ключ шардирования

Четвёртый аргумент Distributed() — выражение шардирования. ClickHouse берёт cityHash64(user_id) % num_shards и направляет строку на нужный шард. Если выражение опущено, используется случайное распределение (round-robin).

Как работает вставка

• При INSERT в Distributed-таблицу координирующая нода разбивает batch по шардам и асинхронно отправляет части на нужные узлы через внутренний TCP.
• Настройка insert_distributed_sync = 1 делает вставку синхронной — нода ждёт подтверждения от всех шардов.
• При асинхронной вставке данные сначала буферизуются в директории data/default/events/ на инициирующей ноде.

Как работает SELECT

Координирующая нода рассылает запрос всем шардам, получает результаты и выполняет финальную агрегацию. При наличии реплик в шарде выбирается одна — по стратегии load_balancing (round_robin, nearest_hostname, random и др.).

-- Проверка распределения данных по шардам
SELECT _shard_num, count()
FROM events
GROUP BY _shard_num
ORDER BY _shard_num;

Подводные камни

• Ключ шардирования должен совпадать с частью ORDER BY — иначе агрегации по этому ключу будут требовать перегонки данных между шардами (distributed GROUP BY становится дорогим).
• Асинхронная вставка через Distributed может потерять данные при падении координирующей ноды — буфер хранится локально, а не реплицируется.
• JOIN между Distributed-таблицами выполняется через GLOBAL JOIN — без него каждый шард выполняет подзапрос независимо, что порождает N*M запросов.
• Перешардирование (изменение числа шардов) — нет встроенного решения: нужно вручную перемещать партиции командой ALTER TABLE MOVE PARTITION.
• Метаданные кластера в config.xml не реплицируются автоматически — конфиг нужно синхронизировать на все ноды вручную или через Ansible/Puppet.
• ON CLUSTER DDL выполняется асинхронно — CREATE TABLE ON CLUSTER может завершиться на части нод, если какая-то недоступна.
• При использовании insert_distributed_sync = 0 (по умолчанию) возможны дубликаты при retry вставки после таймаута.
• Виртуальная колонка _shard_num доступна только при чтении через Distributed — нельзя использовать в локальных таблицах.

• Объяснять distributed tables и sharding как OLTP-механику row-store базы вместо аналитической колоночной модели ClickHouse.
• Путать primary key ClickHouse с уникальным constraint из PostgreSQL или MySQL.
• Игнорировать parts, merges, ORDER BY, sparse index и стоимость маленьких вставок.
• Предлагать синтаксис или транзакционное поведение, которого в ClickHouse нет.

• Кандидат объясняет distributed tables и sharding через реальный механизм ClickHouse, а не общими словами.
• Приводит корректный SQL или диагностический запрос для этой СУБД.
• Называет ограничения, версионные отличия или эксплуатационные последствия.
• Связывает ответ с проектированием приложения, производительностью, надежностью или безопасностью.