Шардирование и распределённые запросы

Кластеры, движок Distributed, балансировка нагрузки и глобальные JOIN

#Результат урока

Вы выберете ключ шардирования по распределению данных и паттернам запросов, измерите перекос и объясните цену fan-out и сетевой пересылки.

#Обзор шардирования

Шардирование — горизонтальное разделение данных между несколькими серверами (шардами).

Клиент
   ↓
┌────────────────────────────────────────────┐
│         Distributed-таблица                │
│         (логическое представление)         │
└────────────────────────────────────────────┘
   ↓              ↓              ↓
┌─────────┐  ┌─────────┐  ┌─────────┐
│ Шард 1  │  │ Шард 2  │  │ Шард 3  │
│ [1/3]   │  │ [2/3]   │  │ [3/3]   │
└─────────┘  └─────────┘  └─────────┘

Преимущества:

Горизонтальное масштабирование записи и чтения
Распределение нагрузки по серверам
Хранение данных больше чем RAM одного сервера

Недостатки:

Сложность администрирования
Глобальные JOIN дороже
Перекос распределения

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

#remote_servers.xml

<clickhouse>
    <remote_servers>
        <cluster_default>
            <!-- Шард 1 с 2 репликами -->
            <shard>
                <internal_replication>true</internal_replication>
                <replica>
                    <host>clickhouse-01-1</host>
                    <port>9000</port>
                </replica>
                <replica>
                    <host>clickhouse-01-2</host>
                    <port>9000</port>
                </replica>
            </shard>
            
            <!-- Шард 2 с 2 репликами -->
            <shard>
                <internal_replication>true</internal_replication>
                <replica>
                    <host>clickhouse-02-1</host>
                    <port>9000</port>
                </replica>
                <replica>
                    <host>clickhouse-02-2</host>
                    <port>9000</port>
                </replica>
            </shard>
            
            <!-- Шард 3 с 2 репликами -->
            <shard>
                <internal_replication>true</internal_replication>
                <replica>
                    <host>clickhouse-03-1</host>
                    <port>9000</port>
                </replica>
                <replica>
                    <host>clickhouse-03-2</host>
                    <port>9000</port>
                </replica>
            </shard>
        </cluster_default>
    </remote_servers>
</clickhouse>

Параметры:

internal_replication — true для репликации внутри ClickHouse
shard — группа серверов с одинаковыми данными
replica — отдельный сервер в шарде

#macros.xml

<clickhouse>
    <macros>
        <!-- Уникально для каждого узла -->
        <cluster>cluster_default</cluster>
        <shard>01</shard>
        <replica>01</replica>
    </macros>
</clickhouse>

#Движок Distributed

#Создание распределённой таблицы

-- Локальная таблица на каждом шарде
CREATE TABLE events_local ON CLUSTER cluster_default
(
    event_time DateTime,
    user_id    UInt64,
    event_type String,
    value      Decimal(10, 2)
)
ENGINE = ReplicatedMergeTree(
    '/clickhouse/tables/{shard}/events',
    '{replica}'
)
ORDER BY (event_time, user_id);

-- Distributed-таблица (логическое представление)
CREATE TABLE events_all ON CLUSTER cluster_default
(
    event_time DateTime,
    user_id    UInt64,
    event_type String,
    value      Decimal(10, 2)
)
ENGINE = Distributed(
    cluster_default,    -- Имя кластера
    default,            -- База данных
    events_local,       -- Локальная таблица
    rand()              -- Шардирующее выражение
);

Параметры Distributed:

cluster — имя кластера из remote_servers.xml
database — БД локальной таблицы
table — имя локальной таблицы
sharding_key — выражение для распределения (опционально)

#Шардирующее выражение

Варианты:

-- Случайное распределение
ENGINE = Distributed(cluster_default, default, events_local, rand());

-- По user_id (равномерное распределение)
ENGINE = Distributed(cluster_default, default, events_local, user_id);

-- По хэшу (более равномерное)
ENGINE = Distributed(cluster_default, default, events_local, sipHash64(user_id));

-- Без ключа (все данные на всех шардах — для репликации)
ENGINE = Distributed(cluster_default, default, events_local);

Рекомендации:

rand() — для равномерного распределения без логики
user_id — для локальности данных пользователя
sipHash64() — для равномерного хэширования

#Вставка данных

#Вставка через Distributed

-- Вставка распределяется по шардам автоматически
INSERT INTO events_all (event_time, user_id, event_type, value)
VALUES 
    (now(), 1, 'click', 1.0),
    (now(), 2, 'view', 2.0),
    (now(), 3, 'purchase', 100.0);

-- Каждая строка попадает на свой шард на основе sharding_key

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

Клиент подключается к любому узлу
Distributed разбирает sharding_key для каждой строки
Строки отправляются на соответствующие шарды
Данные вставляются в local-таблицу

#Вставка напрямую в шард

-- Подключение к конкретному шарду
INSERT INTO events_local ON CLUSTER cluster_default
SELECT * FROM source_data;

-- Или напрямую через клиент
clickhouse-client --host clickhouse-01-1
INSERT INTO events_local VALUES (...);

Преимущества:

Нет накладных расходов на распределение вставки
Контроль над размещением данных

Недостатки:

Нужно управлять подключением к шардам

#Чтение данных

#Запрос через Distributed

-- Запрос выполняется на всех шардах параллельно
SELECT 
    user_id,
    count() AS events
FROM events_all
WHERE event_time >= '2026-03-01'
GROUP BY user_id;

План выполнения:

Distributed отправляет запрос на все шарды
Каждый шард выполняет запрос локально
Результаты агрегируются на координаторе
Клиент получает объединённый результат

#Локальность данных

-- Запрос с фильтром по sharding_key
SELECT * FROM events_all 
WHERE user_id = 123;

-- ClickHouse может оптимизировать и отправить 
-- запрос только на один шард (если sharding_key = user_id)

#WITH TIES для балансировки

-- Ограничение с балансировкой
SELECT * FROM events_all
ORDER BY event_time DESC
LIMIT 100 WITH TIES;

#Глобальные JOIN

#Проблема глобальных JOIN

-- JOIN между Distributed-таблицами
SELECT 
    e.user_id,
    e.event_type,
    u.country
FROM events_all AS e
JOIN users_all AS u ON e.user_id = u.id;

Проблема: Данные могут быть на разных шардах.

#Решения

1. GLOBAL JOIN (данные рассылаются по шардам):

SELECT 
    e.user_id,
    e.event_type,
    u.country
FROM events_all AS e
GLOBAL JOIN users_all AS u ON e.user_id = u.id;

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

Правая таблица (users_all) читается со всех шардов
Данные рассылаются на все шарды
JOIN выполняется локально на каждом шарде
Результаты агрегируются

2. JOIN с локальными таблицами:

-- users_local на каждом шарде
SELECT 
    e.user_id,
    e.event_type,
    u.country
FROM events_all AS e
JOIN users_local AS u ON e.user_id = u.id;

Преимущества:

Нет сетевой передачи
Быстрее чем GLOBAL JOIN

Требование: Таблица users должна быть продублирована на всех шардах.

3. Словари вместо JOIN:

-- Создание словаря
CREATE DICTIONARY users_dict ON CLUSTER cluster_default
(
    user_id UInt64,
    country String
)
PRIMARY KEY user_id
SOURCE(CLICKHOUSE(
    HOST 'localhost'
    PORT 9000
    DB 'default'
    TABLE 'users_local'
))
LAYOUT(HASHED())
LIFETIME(MIN 300 MAX 360);

-- Использование в запросе
SELECT 
    user_id,
    event_type,
    dictGet('users_dict', 'country', user_id) AS country
FROM events_all;

#Балансировка нагрузки

#Распределение записи

-- Вставка через Distributed с rand()
-- Данные равномерно распределяются по шардам
INSERT INTO events_all 
SELECT now(), user_id, 'event', value 
FROM source;

#Распределение чтения

1. round-robin балансировка:

<!-- users.xml -->
<profiles>
    <default>
        <load_balancing>round_robin</load_balancing>
    </default>
</profiles>

2. nearest_hostname:

<profiles>
    <default>
        <load_balancing>nearest_hostname</load_balancing>
    </default>
</profiles>

3. in_order:

<profiles>
    <default>
        <load_balancing>in_order</load_balancing>
    </default>
</profiles>

#Настройки балансировки

-- Для сессии
SET load_balancing = 'round_robin';
SET prefer_localhost_replica = 1;

-- Проверка настроек
SELECT name, value 
FROM system.settings 
WHERE name LIKE 'load_balancing%';

#Мониторинг кластера

#system.clusters

SELECT 
    cluster,
    shard_num,
    replica_num,
    host_name,
    host_address,
    port,
    is_local
FROM system.clusters
WHERE cluster = 'cluster_default';

#system.distributed_ddl_queue

-- Статус выполнения DDL на кластере
SELECT 
    query,
    host,
    shard_num,
    replica_num,
    status,
    exception_code,
    query_finish_time
FROM system.distributed_ddl_queue
ORDER BY entry DESC
LIMIT 20;

#Проверка распределения данных

-- Количество строк по шардам
SELECT 
    _shard_num,
    count() AS rows
FROM events_all
GROUP BY _shard_num
ORDER BY _shard_num;

-- Перекос распределения
SELECT 
    _shard_num,
    count() AS rows,
    count() * 100.0 / sum(count()) OVER () AS pct
FROM events_all
GROUP BY _shard_num;

#Решения, которые нужно обосновать

#1. Выбор ключа шардирования

Хорошо:

-- Равномерное распределение
ENGINE = Distributed(cluster, default, events_local, sipHash64(user_id));

-- Локальность данных пользователя
ENGINE = Distributed(cluster, default, events_local, user_id);

Плохо:

-- Все данные на одном шарде
ENGINE = Distributed(cluster, default, events_local, 1);

-- Неравномерное распределение
ENGINE = Distributed(cluster, default, events_local, toYYYYMM(event_time));

#2. Дублирование справочников

-- users_local на каждом шарде
CREATE TABLE users_local ON CLUSTER cluster_default
(
    user_id UInt64,
    country String
)
ENGINE = ReplicatedMergeTree(...)
ORDER BY user_id;

-- GLOBAL JOIN не нужен
SELECT e.*, u.country 
FROM events_all e 
JOIN users_local u ON e.user_id = u.id;

#3. Избегание GLOBAL JOIN

-- Вместо GLOBAL JOIN используйте словари
SELECT 
    user_id,
    dictGet('users_dict', 'country', user_id) AS country
FROM events_all;

#4. Контроль перекоса

-- Проверка равномерности распределения
SELECT 
    _shard_num,
    count() AS rows,
    round(count() * 100.0 / sum(count()) OVER (), 2) AS pct
FROM events_all
GROUP BY _shard_num
ORDER BY _shard_num;

-- Если pct сильно отличается, пересмотрите ключ шардирования

#5. Агрегация перед объединением

-- Плохо: все данные на координаторе
SELECT * FROM events_all WHERE ...;

-- Хорошо: агрегация на шардах
SELECT 
    shard_id,
    sum(amount) AS shard_total
FROM events_all
GROUP BY shard_id;

#Практика: горячий арендатор

Смоделируйте распределение, где один tenant_id создаёт 40% строк. Сравните минимум два ключа шардирования по строкам, байтам и времени распределённого запроса. Объясните компромисс между равномерностью и локальностью данных одного арендатора; rand() без анализа запросов не считается решением.

На распределённом запросе легко спутать причину и симптом. Дальше разберём EXPLAIN, планы выполнения и способ проверять одну гипотезу за раз.

Клиент ↓ ┌────────────────────────────────────────────┐ │ Distributed-таблица │ │ (логическое представление) │ └────────────────────────────────────────────┘ ↓ ↓ ↓ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ Шард 1 │ │ Шард 2 │ │ Шард 3 │ │ [1/3] │ │ [2/3] │ │ [3/3] │ └─────────┘ └─────────┘ └─────────┘