Стратегия шардирования и Distributed engine

Distributed — виртуальная таблица без данных. Она хранит только конфигурацию маршрутизации: имя кластера, базу данных, имя локальной таблицы и функцию шардирования. При вставке Distributed вычисляет целевой шард для каждой строки и перенаправляет её туда. При SELECT распределяет подзапросы по всем шардам и объединяет результаты.

CREATE TABLE … ENGINE = Distributed

-- Синтаксис Distributed table
-- Параметры: cluster_name, database, local_table, sharding_key_expression
CREATE TABLE events_dist
(
    event_time   DateTime,
    user_id      UInt64,
    event_type   LowCardinality(String),
    properties   String
)
ENGINE = Distributed(
    'mycluster',     -- имя кластера из remote_servers конфига
    'db',            -- база данных где находится локальная таблица
    'events_local',  -- локальная таблица на каждом шарде
    xxHash64(user_id)  -- функция шардирования
);

Distributed table должна быть создана на каждом узле кластера. Схема должна совпадать с локальной таблицей. Распределение строк: shard_num = sharding_key_expression % num_shards.

Стратегии выбора shard key

Shard weights

Shard weights позволяют распределять нагрузку неравномерно между шардами разной ёмкости. Шард с весом 2 получает вдвое больше строк, чем шард с весом 1.

<!-- remote_servers конфиг с shard weights -->
<remote_servers>
    <mycluster>
        <shard>
            <!-- Шард 1: ёмкий сервер -- получает вдвое больше данных -->
            <weight>2</weight>
            <internal_replication>true</internal_replication>
            <replica>
                <host>clickhouse-01</host>
                <port>9000</port>
            </replica>
        </shard>
        <shard>
            <!-- Шард 2: меньший сервер -- получает в два раза меньше данных -->
            <weight>1</weight>
            <internal_replication>true</internal_replication>
            <replica>
                <host>clickhouse-02</host>
                <port>9000</port>
            </replica>
        </shard>
    </mycluster>
</remote_servers>

При весах 2:1 из каждых 3 строк Distributed отправит 2 на шард 1 и 1 на шард 2. Это позволяет использовать серверы разного размера в одном кластере без равного распределения.

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

1. Distributed — виртуальная таблица без данных: хранит только маршрутизационный конфиг и вычисляет shard_num = sharding_key % num_shards для каждой вставки.
2. xxHash64(user_id) — рекомендуемая стратегия для event-данных: обеспечивает равномерное распределение и колокацию данных одного пользователя на одном шарде.
3. rand() даёт максимально равномерное распределение, но исключает colocation — каждый запрос по пользователю читает все шарды.
4. Shard weights позволяют использовать серверы разной ёмкости: шард с <weight>2</weight> получает вдвое больше строк.
5. Distributed table создаётся на каждом узле с такой же схемой, что и локальная таблица.