Перейти к основному содержанию
Движок Distributed в CloudЧтобы создать таблицу с движком Distributed в ClickHouse Cloud, можно использовать табличные функции remote и remoteSecure. Синтаксис Distributed(...) нельзя использовать в ClickHouse Cloud.
Таблицы с движком Distributed не хранят собственные данные, но позволяют выполнять распределённую обработку запросов на нескольких серверах. Чтение автоматически распараллеливается. При чтении используются индексы таблиц на удалённых серверах, если они есть.

​
Создание таблицы

CREATE TABLE [IF NOT EXISTS] [db.]table_name [ON CLUSTER cluster]
(
    name1 [type1] [DEFAULT|MATERIALIZED|ALIAS expr1],
    name2 [type2] [DEFAULT|MATERIALIZED|ALIAS expr2],
    ...
) ENGINE = Distributed(cluster, database, table[, sharding_key[, policy_name]])
[SETTINGS name=value, ...]

​
Из таблицы

Если distributed таблица Distributed указывает на таблицу на текущем сервере, вы можете использовать её схему: 
CREATE TABLE [IF NOT EXISTS] [db.]table_name [ON CLUSTER cluster] AS [db2.]name2 ENGINE = Distributed(cluster, database, table[, sharding_key[, policy_name]]) [SETTINGS name=value, ...]

​
Параметры Distributed

ПараметрОписание
clusterИмя кластера в конфигурационном файле сервера
databaseИмя удаленной базы данных
tableИмя удаленной таблицы
sharding_key (Необязательно)Ключ сегментирования.
Указание sharding_key необходимо в следующих случаях:
  • Для операций INSERT в distributed таблицу (поскольку движку таблицы нужен sharding_key, чтобы определить, как распределить данные). Однако если включена настройка insert_distributed_one_random_shard, то для INSERT ключ сегментирования не требуется.
  • Для использования с optimize_skip_unused_shards, поскольку sharding_key нужен, чтобы определить, какие сегменты следует запрашивать
policy_name (Необязательно)Имя политики; оно будет использоваться для хранения временных файлов при фоновой отправке
См. также

​
Настройки Distributed

НастройкаОписаниеЗначение по умолчанию
fsync_after_insertВыполнять fsync для данных файла после фоновой вставки в Distributed. Гарантирует, что ОС сбросила на диск все вставленные данные на узле-инициаторе.false
fsync_directoriesВыполнять fsync для каталогов. Гарантирует, что ОС обновила метаданные каталогов после операций, связанных с фоновыми вставками в таблицу Distributed (после вставки, после отправки данных в сегмент и т. д.).false
skip_unavailable_shardsЕсли true, ClickHouse молча пропускает недоступные сегменты. Сегмент помечается как недоступный, если: 1) к сегменту невозможно подключиться из-за сбоя соединения. 2) сегмент не разрешается через DNS. 3) таблица отсутствует на сегменте.false
bytes_to_throw_insertЕсли объём сжатых байтов, ожидающих фонового INSERT, превысит это значение, будет сгенерировано исключение. 0 — не генерировать исключение.0
bytes_to_delay_insertЕсли объём сжатых байтов, ожидающих фонового INSERT, превысит это значение, запрос будет задержан. 0 — не задерживать.0
max_delay_to_insertМаксимальная задержка вставки данных в таблицу Distributed в секундах, если для фоновой отправки накопилось много байтов.60
background_insert_batchТо же, что и distributed_background_insert_batch0
background_insert_split_batch_on_failureТо же, что и distributed_background_insert_split_batch_on_failure0
background_insert_sleep_time_msТо же, что и distributed_background_insert_sleep_time_ms0
background_insert_max_sleep_time_msТо же, что и distributed_background_insert_max_sleep_time_ms0
flush_on_detachСбрасывать данные на удалённые узлы при DETACH/DROP/остановке сервера.true
Настройки надёжности хранения (fsync_...):
  • Влияют только на фоновые INSERT (то есть distributed_foreground_insert=false), когда данные сначала сохраняются на диске узла-инициатора, а затем в фоне отправляются в сегменты.
  • Могут значительно снизить производительность INSERT
  • Влияют на запись данных, хранящихся в каталоге distributed-таблицы, на узле, который принял вашу вставку. Если вам нужны гарантии записи данных в базовые таблицы MergeTree, см. настройки надёжности (...fsync...) в system.merge_tree_settings
Для настроек лимитов вставки (..._insert) см. также:
  • настройку distributed_foreground_insert
  • настройку prefer_localhost_replica
  • bytes_to_throw_insert обрабатывается раньше bytes_to_delay_insert, поэтому не следует задавать для него значение меньше, чем bytes_to_delay_insert
Пример 
CREATE TABLE hits_all AS hits
ENGINE = Distributed(logs, default, hits[, sharding_key[, policy_name]])
SETTINGS
    fsync_after_insert=0,
    fsync_directories=0;
Данные будут считываться со всех серверов кластера logs из таблицы default.hits, расположенной на каждом сервере кластера. Данные не только считываются, но и частично обрабатываются на удалённых серверах (насколько это возможно). Например, в запросе с GROUP BY данные будут агрегироваться на удалённых серверах, а промежуточные состояния агрегатных функций будут отправляться на сервер-инициатор запроса. Затем данные будут агрегироваться дальше. Вместо имени базы данных можно использовать константное выражение, возвращающее строку. Например: currentDatabase().

​
Кластеры

Кластеры настраиваются в файле конфигурации сервера: 
<remote_servers>
    <logs>
        <!-- Межсерверный секрет для распределённых запросов на уровне кластера
             по умолчанию: секрет не задан (аутентификация не выполняется)

             Если задан, распределённые запросы будут проверяться на сегментах, то есть как минимум:
             - такой кластер должен существовать на сегменте,
             - такой кластер должен иметь тот же секрет.

             Кроме того (и это важнее), initial_user будет
             использоваться как текущий пользователь для запроса.
        -->
        <!-- <secret></secret> -->
        
        <!-- Необязательный параметр. Разрешены ли распределённые DDL-запросы (предложение ON CLUSTER) для данного кластера. По умолчанию: true (разрешено). -->        
        <!-- <allow_distributed_ddl_queries>true</allow_distributed_ddl_queries> -->
        
        <shard>
            <!-- Необязательный параметр. Вес сегмента при записи данных. По умолчанию: 1. -->
            <weight>1</weight>
            <!-- Необязательный параметр. Имя сегмента. Должно быть непустым и уникальным среди сегментов кластера. Если не указано, будет пустым. -->
            <name>shard_01</name>
            <!-- Необязательный параметр. Записывать ли данные только в одну из реплик. По умолчанию: false (запись данных во все реплики). -->
            <internal_replication>false</internal_replication>
            <replica>
                <!-- Необязательный параметр. Приоритет реплики для балансировки нагрузки (см. также настройку load_balancing). По умолчанию: 1 (меньшее значение означает более высокий приоритет). -->
                <priority>1</priority>
                <host>example01-01-1</host>
                <port>9000</port>
            </replica>
            <replica>
                <host>example01-01-2</host>
                <port>9000</port>
            </replica>
        </shard>
        <shard>
            <weight>2</weight>
            <name>shard_02</name>
            <internal_replication>false</internal_replication>
            <replica>
                <host>example01-02-1</host>
                <port>9000</port>
            </replica>
            <replica>
                <host>example01-02-2</host>
                <secure>1</secure>
                <port>9440</port>
            </replica>
        </shard>
    </logs>
</remote_servers>
Здесь определён кластер с именем logs, состоящий из двух сегментов, каждый из которых содержит две реплики. Сегменты — это серверы, содержащие разные части данных (чтобы прочитать все данные, необходимо обратиться ко всем сегментам). Реплики — это серверы-дубликаты (чтобы прочитать все данные, можно обратиться к данным на любой из реплик). Имена кластеров не должны содержать точек. Для каждого сервера указываются параметры host, port, а также при необходимости user, password, secure, compression, bind_host:
ПараметрОписаниеЗначение по умолчанию
hostАдрес удалённого сервера. Можно использовать либо доменное имя, либо IPv4-адрес или IPv6-адрес. Если указан домен, сервер при запуске выполняет DNS-запрос, и результат сохраняется, пока сервер работает. Если DNS-запрос завершается ошибкой, сервер не запускается. Если вы изменили DNS-запись, перезапустите сервер.-
portTCP-порт для обмена сообщениями (tcp_port в конфигурации, обычно равен 9000). Не следует путать с http_port.-
userИмя пользователя для подключения к удалённому серверу. У этого пользователя должны быть права доступа для подключения к указанному серверу. Доступ настраивается в файле users.xml. Дополнительные сведения см. в разделе Права доступа.default
passwordПароль для подключения к удалённому серверу (не маскируется).”
secureСледует ли использовать защищённое SSL/TLS‑соединение. Обычно также требуется указать порт (порт по умолчанию для защищённого соединения — 9440). Сервер должен прослушивать <tcp_port_secure>9440</tcp_port_secure> и быть настроен с корректными сертификатами.false
compressionИспользовать сжатие данных.true
bind_hostИсходный адрес, который следует использовать при подключении к удалённому серверу с этого узла. Поддерживается только IPv4-адрес. Предназначено для сложных сценариев развертывания, когда необходимо задать исходный IP-адрес, используемый ClickHouse для распределённых запросов.-
При указании реплик для каждого сегмента при чтении будет выбрана одна из доступных реплик. Вы можете настроить алгоритм балансировки нагрузки (предпочтение, к какой реплике обращаться) — см. настройку load_balancing. Если соединение с сервером не удаётся установить, будет предпринята попытка подключения с коротким тайм-аутом. Если подключиться не удалось, будет выбрана следующая реплика, и так для всех реплик. Если попытка подключения не удалась для всех реплик, она будет тем же образом повторена несколько раз. Это повышает устойчивость, но не обеспечивает полной отказоустойчивости: удалённый сервер может принять соединение, но не работать или работать нестабильно. Вы можете указать только один сегмент (в этом случае обработку запросов следует называть remote, а не distributed) или любое количество сегментов. В каждом сегменте можно указать от одной реплики до любого их числа. Для каждого сегмента можно указать разное количество реплик. В конфигурации можно указать столько кластеров, сколько потребуется. Чтобы просмотреть свои кластеры, используйте таблицу system.clusters. Движок Distributed позволяет работать с кластером как с локальным сервером. Однако конфигурацию кластера нельзя задавать динамически, её нужно настраивать в конфигурационном файле сервера. Обычно все серверы в кластере имеют одинаковую конфигурацию кластера (хотя это и не обязательно). Кластеры из конфигурационного файла обновляются на лету, без перезапуска сервера. Если вам нужно каждый раз отправлять запрос неизвестному набору сегментов и реплик, создавать таблицу Distributed не нужно — вместо этого используйте табличную функцию remote. См. раздел Табличные функции.

​
Запись данных

Существует два способа записи данных в кластер: Во-первых, можно определить, на какие серверы какие данные записывать, и выполнять запись напрямую в каждый сегмент. Иными словами, выполнять прямые операторы INSERT в удалённые таблицы кластера, на которые указывает distributed таблица Distributed. Это наиболее гибкое решение, поскольку позволяет использовать любую схему шардирования, даже нетривиальную, если этого требует предметная область. Кроме того, это и наиболее оптимальное решение, так как данные можно записывать в разные сегменты полностью независимо друг от друга. Во-вторых, можно выполнять операторы INSERT в distributed таблицу Distributed. В этом случае таблица сама распределяет вставленные данные по серверам. Чтобы записывать данные в distributed таблицу Distributed, у неё должен быть настроен параметр sharding_key (кроме случая, когда сегмент только один). Для каждого сегмента в конфигурационном файле можно определить <weight>. По умолчанию вес равен 1. Данные распределяются по сегментам в объёме, пропорциональном весу сегмента. Все веса сегментов суммируются, затем вес каждого сегмента делится на общую сумму, чтобы определить долю каждого сегмента. Например, если есть два сегмента, и первый имеет вес 1, а второй — вес 2, то в первый будет отправлена одна треть (1 / 3) вставленных строк, а во второй — две трети (2 / 3). Для каждого сегмента в конфигурационном файле можно определить параметр internal_replication. Если этот параметр установлен в true, операция записи выбирает первую работоспособную реплику и записывает данные в неё. Используйте это, если таблицы, лежащие в основе distributed таблицы Distributed, являются реплицируемыми таблицами (например, используют любой из движков таблиц Replicated*MergeTree). Данные будут записаны в одну из реплик таблицы, а затем автоматически реплицированы на остальные реплики. Если internal_replication установлен в false (значение по умолчанию), данные записываются во все реплики. В этом случае distributed таблица Distributed сама реплицирует данные. Это хуже, чем использование реплицируемых таблиц, поскольку согласованность реплик не проверяется, и со временем они будут содержать немного различающиеся данные. Чтобы выбрать сегмент, в который будет отправлена строка данных, анализируется выражение шардирования, и берётся остаток от деления на общий вес сегментов. Строка отправляется в тот сегмент, которому соответствует полуинтервал остатков от prev_weights до prev_weights + weight, где prev_weights — это общий вес сегментов с меньшими номерами, а weight — вес данного сегмента. Например, если есть два сегмента, и первый имеет вес 9, а второй — вес 10, то строка будет отправлена в первый сегмент для остатков из диапазона [0, 9), а во второй — для остатков из диапазона [9, 19). Выражением шардирования может быть любое выражение из констант и столбцов таблицы, возвращающее целое число. Например, можно использовать выражение rand() для случайного распределения данных или UserID для распределения по остатку от деления идентификатора пользователя (тогда данные одного пользователя будут находиться на одном сегменте, что упрощает выполнение IN и JOIN по пользователям). Если один из столбцов распределён недостаточно равномерно, его можно обернуть в хеш-функцию, например intHash64(UserID). Простой остаток от деления — ограниченное решение для шардирования, и оно подходит не всегда. Оно работает для средних и больших объёмов данных (десятки серверов), но не для очень больших объёмов данных (сотни серверов и более). В последнем случае используйте схему шардирования, подходящую для предметной области, а не таблицы Distributed. О схеме шардирования следует задуматься в следующих случаях:
  • Используются запросы, требующие объединения данных (IN или JOIN) по определённому ключу. Если данные шардированы по этому ключу, можно использовать локальные IN или JOIN вместо GLOBAL IN или GLOBAL JOIN, что значительно эффективнее.
  • Используется большое количество серверов (сотни и более) с большим числом небольших запросов, например запросов к данным отдельных клиентов (например, веб-сайтов, рекламодателей или партнёров). Чтобы небольшие запросы не влияли на весь кластер, имеет смысл размещать данные одного клиента на одном сегменте. Либо можно настроить двухуровневое шардирование: разделить весь кластер на «слои», где слой может состоять из нескольких сегментов. Данные одного клиента располагаются на одном слое, но при необходимости в слой можно добавлять сегменты, а данные внутри них распределяются случайным образом. Таблицы Distributed создаются для каждого слоя, а для глобальных запросов создаётся одна общая distributed таблица.
Данные записываются в фоновом режиме. При вставке в таблицу блок данных просто записывается в локальную файловую систему. Затем данные как можно скорее отправляются на удалённые серверы в фоновом режиме. Периодичность отправки данных задаётся настройками distributed_background_insert_sleep_time_ms и distributed_background_insert_max_sleep_time_ms. Движок Distributed отправляет каждый файл со вставленными данными отдельно, но вы можете включить батч-отправку файлов с помощью настройки distributed_background_insert_batch. Эта настройка повышает производительность кластера за счёт более эффективного использования ресурсов локального сервера и сети. Следует проверять, что данные отправляются успешно, просматривая список файлов (данных, ожидающих отправки) в каталоге таблицы: /var/lib/clickhouse/data/database/table/. Количество потоков, выполняющих фоновые задачи, можно задать с помощью настройки background_distributed_schedule_pool_size. Если после INSERT в distributed таблицу Distributed сервер вышел из строя или был аварийно перезапущен (например, из-за аппаратного сбоя), вставленные данные могут быть потеряны. Если в каталоге таблицы обнаружена повреждённая часть данных, она переносится в подкаталог broken и больше не используется.

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

При выполнении запроса к distributed таблице Distributed запросы SELECT отправляются во все сегменты и работают независимо от того, как данные распределены между сегментами (они могут быть распределены совершенно случайным образом). При добавлении нового сегмента переносить в него старые данные не требуется. Вместо этого можно записывать в него новые данные, задав больший вес: данные будут распределены немного неравномерно, но запросы продолжат работать корректно и эффективно. Когда включена опция max_parallel_replicas, обработка запроса распараллеливается между всеми репликами в пределах одного сегмента. Дополнительные сведения см. в разделе max_parallel_replicas. Чтобы узнать больше о том, как обрабатываются распределенные запросы in и global in, см. эту документацию.

​
Виртуальные столбцы

​
_Shard_num

_shard_num — содержит значение shard_num из таблицы system.clusters. Тип: UInt32.
Поскольку табличные функции remote и [cluster](../../../sql-reference/table-functions/cluster.md) внутренне создают временную distributed таблицу, _shard_num` также доступен и в них.
См. также
Последнее изменение 10 июня 2026 г.