Перейти к основному содержанию

​
Введение

ClickHouse обрабатывает запросы чрезвычайно быстро, но как эти запросы распределяются между несколькими серверами и выполняются параллельно?
В этом руководстве мы сначала рассмотрим, как ClickHouse распределяет запрос по нескольким сегментам с помощью distributed таблиц, а затем — как запрос может задействовать несколько реплик при выполнении.

​
Архитектура с сегментированием

В архитектуре shared-nothing кластер обычно разделяется на несколько сегментов, при этом каждый сегмент содержит подмножество всех данных. Над этими сегментами располагается distributed таблица, предоставляющая единое представление всего набора данных. Операции чтения можно направлять в локальную таблицу. В этом случае выполнение запроса будет происходить только на указанном сегменте. Либо запрос можно направить в distributed таблицу, и тогда каждый сегмент выполнит данный запрос. Сервер, на котором был выполнен запрос к distributed таблице, агрегирует данные и вернет ответ клиенту: На рисунке выше показано, что происходит, когда клиент выполняет запрос к distributed таблице:
  1. Запрос SELECT отправляется в distributed таблицу на одном из узлов произвольным образом (по стратегии round-robin или после маршрутизации на конкретный сервер с помощью балансировщика нагрузки). Этот узел начинает выступать в роли координатора.
  2. Узел определяет каждый сегмент, который должен выполнить запрос, на основе информации, указанной в distributed таблице, после чего запрос отправляется в каждый сегмент.
  3. Каждый сегмент читает, фильтрует и агрегирует данные локально, а затем отправляет координатору состояние, пригодное для слияния.
  4. Координирующий узел объединяет данные, а затем отправляет ответ клиенту.
Когда в эту схему добавляются реплики, процесс остается практически таким же, с той лишь разницей, что запрос выполняет только одна реплика из каждого сегмента. Это означает, что больше запросов можно обрабатывать параллельно.

​
Архитектура без сегментирования

Архитектура ClickHouse Cloud значительно отличается от представленной выше. (Подробнее см. в разделе “Архитектура ClickHouse Cloud”). Благодаря разделению вычислительных ресурсов и хранилища, а также практически неограниченному объему хранения, потребность в сегментах становится менее важной. На рисунке ниже показана архитектура ClickHouse Cloud: Эта архитектура позволяет почти мгновенно добавлять и удалять реплики, обеспечивая очень высокую масштабируемость кластера. Кластер ClickHouse Keeper (показан справа) обеспечивает единый источник истины для метаданных. Реплики могут получать метаданные из кластера ClickHouse Keeper и поддерживать одни и те же данные. Сами данные хранятся в объектном хранилище, а SSD-кэш позволяет ускорять запросы. Но как теперь распределяется выполнение запросов между несколькими серверами? В архитектуре с сегментированием это было вполне очевидно, поскольку каждый сегмент мог выполнять запрос на подмножестве данных. Как это работает, когда сегментирования нет?

​
Введение в параллельные реплики

Чтобы распараллелить выполнение запроса между несколькими серверами, сначала нужно иметь возможность назначить один из серверов координатором. Координатор — это узел, который формирует список задач для выполнения, следит за тем, чтобы все они были выполнены и агрегированы, а затем возвращает результат клиенту. Как и в большинстве распределенных систем, эту роль выполняет узел, который получает исходный запрос. Также нужно определить единицу работы. В архитектуре с сегментами единицей работы является сегмент — подмножество данных. В случае параллельных реплик в качестве единицы работы используется небольшая часть таблицы — гранулы. Теперь посмотрим, как это работает на практике, с помощью схемы ниже: С параллельными репликами:
  1. Запрос от клиента проходит через балансировщик нагрузки и отправляется на один из узлов. Этот узел становится координатором для данного запроса.
  2. Узел анализирует индекс каждой части и выбирает подходящие части и гранулы для обработки.
  3. Координатор разбивает рабочую нагрузку на набор гранул, которые могут быть назначены разным репликам.
  4. Каждый набор гранул обрабатывается соответствующими репликами, а по завершении на координатор отправляется состояние, пригодное для слияния.
  5. Наконец, координатор объединяет все результаты от реплик и возвращает ответ клиенту.
Описанные выше шаги показывают, как параллельные реплики работают в теории. Однако на практике есть множество факторов, которые могут помешать такой схеме работать идеально:
  1. Некоторые реплики могут быть недоступны.
  2. Репликация в ClickHouse асинхронна, поэтому в определенный момент времени у реплик могут быть разные части.
  3. Нужно как-то учитывать высокую задержку на отдельных репликах.
  4. Файловый кэш различается от реплики к реплике в зависимости от активности на каждой из них, поэтому случайное назначение задач может привести к менее оптимальной производительности из-за локальности кэша.
Как именно удается справиться с этими факторами, мы рассмотрим в следующих разделах.

​
Уведомления

Чтобы решить проблемы (1) и (2) из приведенного выше списка, мы ввели понятие уведомления. Давайте посмотрим, как это работает, на схеме ниже:
  1. Запрос от клиента проходит через балансировщик нагрузки и отправляется на один из узлов. Этот узел становится координатором данного запроса.
  2. Узел-координатор отправляет запрос на получение уведомлений от всех реплик в кластере. У реплик могут быть немного разные представления о текущем наборе частей таблицы. Поэтому эту информацию нужно собрать, чтобы избежать неверных решений при планировании.
  3. Затем узел-координатор использует уведомления, чтобы определить набор гранул, которые можно назначить разным репликам. Здесь, например, видно, что реплике 2 не была назначена ни одна гранула из части 3, потому что эта реплика не включила эту часть в свое уведомление. Также обратите внимание, что реплике 3 не было назначено ни одной задачи, потому что от этой реплики не поступило уведомление.
  4. После того как каждая реплика обработала запрос на своем подмножестве гранул и объединяемое состояние было отправлено обратно координатору, координатор объединяет результаты, и ответ отправляется клиенту.

​
Динамическая координация

Чтобы решить проблему хвостовой задержки, мы добавили динамическую координацию. Это означает, что все гранулы не отправляются реплике одним запросом; вместо этого каждая реплика может запрашивать у координатора новую задачу (набор гранул для обработки). Координатор будет выдавать реплике набор гранул на основе полученного уведомления. Предположим, что мы находимся на этапе, когда все реплики уже отправили уведомление со всеми частями. На рисунке ниже показано, как работает динамическая координация:
  1. Реплики сообщают узлу-координатору, что готовы обрабатывать задачи; они также могут указать, какой объём работы способны взять.
  2. Координатор назначает репликам задачи.
  1. Реплики 1 и 2 завершают свою задачу очень быстро. Они запрашивают у узла-координатора ещё одну задачу.
  2. Координатор назначает новые задачи репликам 1 и 2.
  1. Все реплики уже завершили обработку своих задач. Они запрашивают дополнительные задачи.
  2. Координатор, используя уведомления, проверяет, какие задачи ещё осталось обработать, но таких задач больше нет.
  3. Координатор сообщает репликам, что всё обработано. Теперь он объединит все состояния, подлежащие слиянию, и вернёт ответ на запрос.

​
Управление локальностью кэша

Последний потенциальный вопрос — как обеспечить локальность кэша. Если запрос выполняется несколько раз, как гарантировать, что одна и та же задача будет направляться на ту же реплику? В предыдущем примере задачи были распределены так:
Реплика 1Реплика 2Реплика 3
Часть 1g1, g6, g7g2, g4, g5g3
Часть 2g1g2, g4, g5g3
Часть 3g1, g6g2, g4, g5g3
Чтобы одни и те же задачи назначались одним и тем же репликам и могли использовать кэш, происходят две вещи. Вычисляется хеш part + набора гранул (то есть задачи). Затем для назначения задачи берется остаток от деления на число реплик. В теории это звучит хорошо, но на практике внезапная нагрузка на одну реплику или ухудшение состояния сети могут увеличить хвостовую задержку, если одна и та же реплика постоянно используется для выполнения определенных задач. Если max_parallel_replicas меньше числа реплик, то для выполнения запроса случайным образом выбираются реплики.

​
Перехват задач

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

​
Ограничения

У этой возможности есть известные ограничения; основные из них описаны в этом разделе.
Если вы столкнулись с проблемой, которая не относится ни к одному из перечисленных ниже ограничений, и подозреваете, что её причина — параллельная реплика, пожалуйста, создайте issue в GitHub, используя метку comp-parallel-replicas.

​
Исследование проблем с параллельными репликами

Вы можете проверить, какие настройки используются для каждого запроса, в таблице system.query_log. Также можно посмотреть таблицу system.events, чтобы увидеть все события, произошедшие на сервере, а также использовать табличную функцию clusterAllReplicas, чтобы увидеть таблицы на всех репликах (если вы используете Cloud, укажите default).
Query
Response
Таблица system.text_log также содержит информацию о выполнении запросов при использовании параллельных реплик:
Query
Response
Наконец, вы также можете использовать EXPLAIN PIPELINE. Он показывает, как ClickHouse будет выполнять запрос и какие ресурсы будут задействованы при его выполнении. Возьмем, к примеру, следующий запрос:
Давайте посмотрим на конвейер выполнения запроса без параллельной реплики:
EXPLAIN PIPELINE (without parallel replica)
А теперь — с параллельной репликой:
EXPLAIN PIPELINE (with parallel replica)
Последнее изменение 2 июля 2026 г.