> ## Documentation Index
> Fetch the complete documentation index at: https://private-7c7dfe99-fix-nav-issues.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.
> Таблицы с движком Distributed не хранят собственные данные, но позволяют выполнять распределённую обработку запросов на нескольких серверах. Чтение автоматически распараллеливается. При чтении используются индексы таблиц на удалённых серверах, если они есть.
# Движок таблицы Distributed
**Движок Distributed в Cloud**
Чтобы создать таблицу с движком Distributed в ClickHouse Cloud, можно использовать [табличные функции `remote` и `remoteSecure`](/ru/reference/functions/table-functions/remote).
Синтаксис `Distributed(...)` нельзя использовать в ClickHouse Cloud.
Таблицы с движком Distributed не хранят собственные данные, но позволяют выполнять распределённую обработку запросов на нескольких серверах.
Чтение автоматически распараллеливается. При чтении используются индексы таблиц на удалённых серверах, если они есть.
## Создание таблицы
```sql theme={null}
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` указывает на таблицу на текущем сервере, вы можете использовать её схему:
```sql theme={null}
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\_foreground\_insert](/ru/reference/settings/session-settings#distributed_foreground_insert)
* [MergeTree](/ru/reference/engines/table-engines/mergetree-family/mergetree#table_engine-mergetree-multiple-volumes) с примерами
### Настройки 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_batch`](/ru/reference/settings/session-settings#distributed_background_insert_batch) | `0` |
| `background_insert_split_batch_on_failure` | То же, что и [`distributed_background_insert_split_batch_on_failure`](/ru/reference/settings/session-settings#distributed_background_insert_split_batch_on_failure) | `0` |
| `background_insert_sleep_time_ms` | То же, что и [`distributed_background_insert_sleep_time_ms`](/ru/reference/settings/session-settings#distributed_background_insert_sleep_time_ms) | `0` |
| `background_insert_max_sleep_time_ms` | То же, что и [`distributed_background_insert_max_sleep_time_ms`](/ru/reference/settings/session-settings#distributed_background_insert_max_sleep_time_ms) | `0` |
| `flush_on_detach` | Сбрасывать данные на удалённые узлы при `DETACH`/`DROP`/остановке сервера. | `true` |
**Настройки надёжности хранения** (`fsync_...`):
* Влияют только на фоновые `INSERT` (то есть `distributed_foreground_insert=false`), когда данные сначала сохраняются на диске узла-инициатора, а затем в фоне отправляются в сегменты.
* Могут значительно снизить производительность `INSERT`
* Влияют на запись данных, хранящихся в каталоге distributed-таблицы, на **узле, который принял вашу вставку**. Если вам нужны гарантии записи данных в базовые таблицы MergeTree, см. настройки надёжности (`...fsync...`) в `system.merge_tree_settings`
Для **настроек лимитов вставки** (`..._insert`) см. также:
* настройку [`distributed_foreground_insert`](/ru/reference/settings/session-settings#distributed_foreground_insert)
* настройку [`prefer_localhost_replica`](/ru/reference/settings/session-settings#prefer_localhost_replica)
* `bytes_to_throw_insert` обрабатывается раньше `bytes_to_delay_insert`, поэтому не следует задавать для него значение меньше, чем `bytes_to_delay_insert`
**Пример**
```sql theme={null}
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()`.
## Кластеры
Кластеры настраиваются в [файле конфигурации сервера](/ru/concepts/features/configuration/server-config/configuration-files):
```xml theme={null}
1
shard_01
false
1
example01-01-1
9000
example01-01-2
9000
2
shard_02
false
example01-02-1
9000
example01-02-2
1
9440
```
Здесь определён кластер с именем `logs`, состоящий из двух сегментов, каждый из которых содержит две реплики. Сегменты — это серверы, содержащие разные части данных (чтобы прочитать все данные, необходимо обратиться ко всем сегментам). Реплики — это серверы-дубликаты (чтобы прочитать все данные, можно обратиться к данным на любой из реплик).
Имена кластеров не должны содержать точек.
Для каждого сервера указываются параметры `host`, `port`, а также при необходимости `user`, `password`, `secure`, `compression`, `bind_host`:
| Параметр | Описание | Значение по умолчанию |
| ------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | --------------------- |
| `host` | Адрес удалённого сервера. Можно использовать либо доменное имя, либо IPv4-адрес или IPv6-адрес. Если указан домен, сервер при запуске выполняет DNS-запрос, и результат сохраняется, пока сервер работает. Если DNS-запрос завершается ошибкой, сервер не запускается. Если вы изменили DNS-запись, перезапустите сервер. | - |
| `port` | TCP-порт для обмена сообщениями (`tcp_port` в конфигурации, обычно равен 9000). Не следует путать с `http_port`. | - |
| `user` | Имя пользователя для подключения к удалённому серверу. У этого пользователя должны быть права доступа для подключения к указанному серверу. Доступ настраивается в файле `users.xml`. Дополнительные сведения см. в разделе [Права доступа](/ru/concepts/features/security/access-rights). | `default` |
| `password` | Пароль для подключения к удалённому серверу (не маскируется). | '' |
| `secure` | Следует ли использовать защищённое SSL/TLS‑соединение. Обычно также требуется указать порт (порт по умолчанию для защищённого соединения — `9440`). Сервер должен прослушивать `9440` и быть настроен с корректными сертификатами. | `false` |
| `compression` | Использовать сжатие данных. | `true` |
| `bind_host` | Исходный адрес, который следует использовать при подключении к удалённому серверу с этого узла. Поддерживается только IPv4-адрес. Предназначено для сложных сценариев развертывания, когда необходимо задать исходный IP-адрес, используемый ClickHouse для распределённых запросов. | - |
При указании реплик для каждого сегмента при чтении будет выбрана одна из доступных реплик. Вы можете настроить алгоритм балансировки нагрузки (предпочтение, к какой реплике обращаться) — см. настройку [load\_balancing](/ru/reference/settings/session-settings#load_balancing). Если соединение с сервером не удаётся установить, будет предпринята попытка подключения с коротким тайм-аутом. Если подключиться не удалось, будет выбрана следующая реплика, и так для всех реплик. Если попытка подключения не удалась для всех реплик, она будет тем же образом повторена несколько раз. Это повышает устойчивость, но не обеспечивает полной отказоустойчивости: удалённый сервер может принять соединение, но не работать или работать нестабильно.
Вы можете указать только один сегмент (в этом случае обработку запросов следует называть remote, а не distributed) или любое количество сегментов. В каждом сегменте можно указать от одной реплики до любого их числа. Для каждого сегмента можно указать разное количество реплик.
В конфигурации можно указать столько кластеров, сколько потребуется.
Чтобы просмотреть свои кластеры, используйте таблицу `system.clusters`.
Движок `Distributed` позволяет работать с кластером как с локальным сервером. Однако конфигурацию кластера нельзя задавать динамически, её нужно настраивать в конфигурационном файле сервера. Обычно все серверы в кластере имеют одинаковую конфигурацию кластера (хотя это и не обязательно). Кластеры из конфигурационного файла обновляются на лету, без перезапуска сервера.
Если вам нужно каждый раз отправлять запрос неизвестному набору сегментов и реплик, создавать таблицу `Distributed` не нужно — вместо этого используйте табличную функцию `remote`. См. раздел [Табличные функции](/ru/reference/functions/table-functions).
## Запись данных
Существует два способа записи данных в кластер:
Во-первых, можно определить, на какие серверы какие данные записывать, и выполнять запись напрямую в каждый сегмент. Иными словами, выполнять прямые операторы `INSERT` в удалённые таблицы кластера, на которые указывает distributed таблица `Distributed`. Это наиболее гибкое решение, поскольку позволяет использовать любую схему шардирования, даже нетривиальную, если этого требует предметная область. Кроме того, это и наиболее оптимальное решение, так как данные можно записывать в разные сегменты полностью независимо друг от друга.
Во-вторых, можно выполнять операторы `INSERT` в distributed таблицу `Distributed`. В этом случае таблица сама распределяет вставленные данные по серверам. Чтобы записывать данные в distributed таблицу `Distributed`, у неё должен быть настроен параметр `sharding_key` (кроме случая, когда сегмент только один).
Для каждого сегмента в конфигурационном файле можно определить ``. По умолчанию вес равен `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](/ru/reference/settings/session-settings#distributed_background_insert_sleep_time_ms) и [distributed\_background\_insert\_max\_sleep\_time\_ms](/ru/reference/settings/session-settings#distributed_background_insert_max_sleep_time_ms). Движок `Distributed` отправляет каждый файл со вставленными данными отдельно, но вы можете включить батч-отправку файлов с помощью настройки [distributed\_background\_insert\_batch](/ru/reference/settings/session-settings#distributed_background_insert_batch). Эта настройка повышает производительность кластера за счёт более эффективного использования ресурсов локального сервера и сети. Следует проверять, что данные отправляются успешно, просматривая список файлов (данных, ожидающих отправки) в каталоге таблицы: `/var/lib/clickhouse/data/database/table/`. Количество потоков, выполняющих фоновые задачи, можно задать с помощью настройки [background\_distributed\_schedule\_pool\_size](/ru/reference/settings/server-settings/settings#background_distributed_schedule_pool_size).
Если после `INSERT` в distributed таблицу `Distributed` сервер вышел из строя или был аварийно перезапущен (например, из-за аппаратного сбоя), вставленные данные могут быть потеряны. Если в каталоге таблицы обнаружена повреждённая часть данных, она переносится в подкаталог `broken` и больше не используется.
## Чтение данных
При выполнении запроса к distributed таблице `Distributed` запросы `SELECT` отправляются во все сегменты и работают независимо от того, как данные распределены между сегментами (они могут быть распределены совершенно случайным образом). При добавлении нового сегмента переносить в него старые данные не требуется. Вместо этого можно записывать в него новые данные, задав больший вес: данные будут распределены немного неравномерно, но запросы продолжат работать корректно и эффективно.
Когда включена опция `max_parallel_replicas`, обработка запроса распараллеливается между всеми репликами в пределах одного сегмента. Дополнительные сведения см. в разделе [max\_parallel\_replicas](/ru/reference/settings/session-settings#max_parallel_replicas).
Чтобы узнать больше о том, как обрабатываются распределенные запросы `in` и `global in`, см. [эту документацию](/ru/reference/statements/in#distributed-subqueries).
## Виртуальные столбцы
#### \_Shard\_num
`_shard_num` — содержит значение `shard_num` из таблицы `system.clusters`. Тип: [UInt32](/ru/reference/data-types/int-uint).
Поскольку табличные функции [`remote`](/ru/reference/functions/table-functions/remote) и \[`cluster](../../../sql-reference/table-functions/cluster.md) внутренне создают временную distributed таблицу, `\_shard\_num\` также доступен и в них.
**См. также**
* [Описание виртуальных столбцов](/ru/reference/engines/table-engines#table_engines-virtual_columns)
* настройка [`background_distributed_schedule_pool_size`](/ru/reference/settings/server-settings/settings#background_distributed_schedule_pool_size)
* функции [`shardNum()`](/ru/reference/functions/regular-functions/other-functions#shardNum) и [`shardCount()`](/ru/reference/functions/regular-functions/other-functions#shardCount)