KB/базы данныхdatabases/колоночные / olapcolumnar / olap/ClickHouse/настройка шардовsharding

ClickHouse — от одного узла к кластеруClickHouse — from one node to a cluster

Настройка шардовSharding

Репликация (см. архитектуру) даёт отказоустойчивость, но не масштабирует запись и объём данных — для этого нужно шардирование. Здесь — как реально описывается топология кластера, что такое `Distributed`-таблица и как выбор ключа шардирования решает, будет ли нагрузка распределена равномерно или соберётся на одном "горячем" узле.Replication (see architecture) gives fault tolerance but doesn't scale writes or data volume — that needs sharding. Here's how cluster topology is actually described, what a `Distributed` table is, and how the sharding-key choice decides whether load spreads evenly or piles up on one "hot" node.

частый вопросcommon interview question

Топология кластераCluster topology

Кластер описывается конфигом с топологией (`remote_servers` — секция в `config.d/`, список именованных кластеров, в каждом список шардов, в каждом шарде список реплик). Это чисто декларативное описание "какие узлы существуют и как они группируются" — сам ClickHouse не занимается автообнаружением узлов или динамическим ребалансированием, как это делают некоторые NoSQL-кластеры; топология статична, пока её не поменяют вручную.A cluster is described by a config with topology (remote_servers — a section in config.d/, a list of named clusters, each with a list of shards, each shard with a list of replicas). This is a purely declarative description of "which nodes exist and how they group" — ClickHouse itself doesn't do node auto-discovery or dynamic rebalancing the way some NoSQL clusters do; the topology is static until changed by hand.

<remote_servers>
    <trades_cluster>
        <shard>
            <replica><host>ch1</host><port>9000</port></replica>
            <replica><host>ch2</host><port>9000</port></replica>
        </shard>
        <shard>
            <replica><host>ch3</host><port>9000</port></replica>
            <replica><host>ch4</host><port>9000</port></replica>
        </shard>
    </trades_cluster>
</remote_servers>

Distributed-таблицаThe Distributed table

`Distributed` — движок-обёртка, которая сама не хранит данные, а маршрутизирует запросы к нижележащим локальным таблицам (обычно `ReplicatedMergeTree`) на каждом узле шарда и объединяет результат. `INSERT` в `Distributed`-таблицу без ключа шардирования уходит на случайный шард; с ключом — вычисляется через `sharding_key`, чтобы связанные строки предсказуемо попадали на один и тот же узел.Distributed is a wrapper engine that doesn't store data itself but routes queries to the underlying local tables (typically ReplicatedMergeTree) on each shard node and merges the result. An INSERT into a Distributed table without a sharding key goes to a random shard; with one, it's computed via sharding_key so related rows land predictably on the same node.

CREATE TABLE trades_local ON CLUSTER trades_cluster (
    ts     DateTime64(3),
    symbol LowCardinality(String),
    price  Float64,
    qty    Float64
) ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/trades', '{replica}')
ORDER BY (symbol, ts);

CREATE TABLE trades ON CLUSTER trades_cluster
AS trades_local
ENGINE = Distributed(trades_cluster, currentDatabase(), trades_local, cityHash64(symbol));

Выбор ключа шардированияChoosing the sharding key

Ключ шардирования (`sharding_key` в `Distributed`) определяет, насколько равномерно распределятся и нагрузка на запись, и объём данных между узлами. Плохой ключ — с низкой кардинальностью или сильным перекосом (например, один доминирующий символ на бирже, если шардировать по `symbol` без хэширования) — создаёт "горячий" шард с непропорциональной долей трафика, обесценивая саму идею горизонтального масштабирования. `cityHash64(column)` — типичный способ равномерно "размазать" даже сильно неравномерно распределённые значения по шардам.The sharding key (sharding_key in Distributed) determines how evenly both write load and data volume spread across nodes. A poor key — low cardinality or heavily skewed (say, one dominant exchange symbol if sharding by symbol without hashing) — creates a "hot" shard carrying a disproportionate share of traffic, undermining the whole point of horizontal scaling. cityHash64(column) is a typical way to spread even heavily skewed values evenly across shards.

Балансировка чтения между репликами внутри шарда настраивается отдельно (`load_balancing`: `random` — по умолчанию, равномерно; `nearest_hostname` — предпочитает реплику с ближайшим по имени хостом, полезно при физическом разнесении по дата-центрам; `in_order` — всегда первую доступную по списку). Для запросов, чувствительных к консистентности сразу после записи, стоит явно читать с той же реплики, куда шла запись, а не полагаться на случайный выбор.Read balancing across replicas within a shard is configured separately (load_balancing: random — the default, even; nearest_hostname — prefers the replica with the closest-named host, useful when physically spread across data centers; in_order — always the first available in the list). For queries sensitive to consistency right after a write, it's worth explicitly reading from the same replica the write went to, rather than relying on random selection.

ГраблиGotchas

ON CLUSTER не гарантирует атомарность на всех узлахdoesn't guarantee atomicity across all nodes

`CREATE TABLE ... ON CLUSTER` рассылает DDL на все узлы, но выполнение на каждом узле независимо — если один узел временно недоступен в момент выполнения, он получит команду из очереди позже (через тот же механизм репликации метаданных, что и куски), но между этим моментом и синхронизацией схема кластера временно рассогласована. На проде это означает: после `ON CLUSTER` DDL стоит проверить, что схема применилась одинаково на всех узлах, а не считать операцию завершённой сразу после ответа.CREATE TABLE ... ON CLUSTER broadcasts DDL to every node, but execution on each node is independent — if a node is temporarily unavailable at execution time, it picks the command up later from a queue (via the same metadata-replication mechanism as parts), but between that moment and sync, the cluster schema is temporarily inconsistent. In production this means: after an ON CLUSTER DDL, verify the schema applied identically everywhere rather than treating the operation as done as soon as it returns.

Cross-shard JOIN/GROUP BY требует пересылки данных между узламиCross-shard JOIN/GROUP BY requires shuffling data between nodes

Запрос через `Distributed`-таблицу с `GROUP BY` по колонке, не совпадающей с ключом шардирования, не может агрегировать полностью локально на каждом шарде — часть данных должна пересылаться по сети между узлами для финальной агрегации (`GLOBAL JOIN`/`distributed_group_by_no_merge` — механизмы, явно управляющие этим). Это существенно медленнее, чем агрегация по ключу шардирования, и на больших кластерах может насытить сеть — проектирование ключа шардирования должно учитывать не только равномерность, но и то, какие агрегации выполняются чаще всего.A query through a Distributed table with a GROUP BY on a column that doesn't match the sharding key can't aggregate fully locally on each shard — some data must be shuffled over the network between nodes for the final aggregation (GLOBAL JOIN/distributed_group_by_no_merge are mechanisms that explicitly control this). This is substantially slower than aggregating by the sharding key, and on large clusters can saturate the network — designing the sharding key needs to account not just for evenness but for which aggregations run most often.