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

Движок распределенных таблиц

осторожно

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

Таблицы с движком Distributed не хранят собственных данных, но позволяют распределенную обработку запросов на нескольких серверах. Чтение автоматически распараллеливается. Во время чтения используются индексы таблиц на удаленных серверах, если таковые имеются.

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

Из таблицы​

Когда таблица Distributed ссылается на таблицу на текущем сервере, вы можете принять схему этой таблицы:

Параметры Distributed​

cluster​

cluster - имя кластера в файле конфигурации сервера

database​

database - имя удаленной базы данных

table​

table - имя удаленной таблицы

sharding_key​

sharding_key - (по желанию) ключ шардирования

Указание sharding_key необходимо для следующего:

  • Для INSERT в распределенную таблицу (так как движок таблицы нуждается в sharding_key, чтобы определить, как разделить данные). Однако, если включена настройка insert_distributed_one_random_shard, то INSERT не требует ключа шардирования.
  • Для использования с optimize_skip_unused_shards, поскольку sharding_key необходим для определения, какие шардов следует опрашивать.

policy_name​

policy_name - (по желанию) имя политики, оно будет использоваться для хранения временных файлов для фоновых отправок.

Смотрите также

Настройки Distributed​

fsync_after_insert​

fsync_after_insert - выполнить fsync для данных файла после фонового вставки в Distributed. Гарантирует, что ОС сбросила все вставленные данные на диск инициирующего узла.

fsync_directories​

fsync_directories - выполнить fsync для директорий. Гарантирует, что ОС обновила метаданные директории после операций, связанных с фоновыми вставками в таблицу Distributed (после вставки, после отправки данных в шард и т.д.).

skip_unavailable_shards​

skip_unavailable_shards - Если true, ClickHouse молча пропускает недоступные шардов. Шард отмечен как недоступный, когда: 1) к шард нельзя подключиться из-за сбоя соединения. 2) Шард неразрешим через DNS. 3) Таблица не существует на шарде. По умолчанию false.

bytes_to_throw_insert​

bytes_to_throw_insert - если более чем это количество сжатых байт будет ожидать фоновой вставки, будет выброшено исключение. 0 - не выбрасывать. По умолчанию 0.

bytes_to_delay_insert​

bytes_to_delay_insert - если более чем это количество сжатых байт будет ожидать фоновой вставки, запрос будет задержан. 0 - не задерживать. По умолчанию 0.

max_delay_to_insert​

max_delay_to_insert - максимальная задержка вставки данных в таблицу Distributed в секундах, если есть много ожидающих байт для фоновой отправки. По умолчанию 60.

background_insert_batch​

background_insert_batch - то же самое, что и distributed_background_insert_batch

background_insert_split_batch_on_failure​

background_insert_split_batch_on_failure - то же самое, что и distributed_background_insert_split_batch_on_failure

background_insert_sleep_time_ms​

background_insert_sleep_time_ms - то же самое, что и distributed_background_insert_sleep_time_ms

background_insert_max_sleep_time_ms​

background_insert_max_sleep_time_ms - то же самое, что и distributed_background_insert_max_sleep_time_ms

flush_on_detach​

flush_on_detach - Сбросить данные на удаленные узлы при DETACH/DROP/выключении сервера. По умолчанию true.

примечание

Настройки долговечности (fsync_...):

  • Влияют только на фоновые вставки (т.е. distributed_foreground_insert=false), когда данные сначала хранятся на диске инициирующего узла и позже, в фоновом режиме, отправляются на шард.
  • Могут значительно снизить производительность вставок
  • Влияют на запись данных, хранящихся внутри папки таблицы 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.

Пример

Данные будут считываться со всех серверов в кластере logs, из таблицы default.hits, расположенной на каждом сервере в кластере. Данные не только считываются, но и частично обрабатываются на удаленных серверах (насколько это возможно). Например, для запроса с GROUP BY данные будут агрегированы на удаленных серверах, а промежуточные состояния агрегатных функций будут отправлены на сервер запроса. Затем данные будут дополнительно агрегированы.

Вместо имени базы данных вы можете использовать константное выражение, которое возвращает строку. Например: currentDatabase().

Кластеры​

Кластеры настраиваются в файле конфигурации сервера:

Здесь кластер с именем logs, который состоит из двух шардов, каждый из которых содержит две реплики. Шарды ссылаются на серверы, которые содержат разные части данных (чтобы прочитать все данные, необходимо получить доступ ко всем шардом). Реплики - это дублирующие серверы (чтобы прочитать все данные, вы можете получить доступ к данным на любой из реплик).

Имена кластеров не должны содержать точек.

Для каждого сервера указываются параметры host, port, и, при желании, user, password, secure, compression:

  • host – адрес удаленного сервера. Вы можете использовать либо домен, либо адрес IPv4 или IPv6. Если вы указываете домен, сервер выполняет DNS-запрос при запуске, и результат сохраняется, пока сервер работает. Если DNS-запрос не удался, сервер не запускается. Если вы измените DNS-запись, перезапустите сервер.
  • port – TCP-порт для деятельности мессенджера (tcp_port в конфигурации, обычно установлен на 9000). Не путать с http_port.
  • user – имя пользователя для подключения к удаленному серверу. Значение по умолчанию – пользователь default. Этот пользователь должен иметь доступ для подключения к указанному серверу. Доступ настраивается в файле users.xml. Для получения дополнительной информации см. раздел Права доступа.
  • password – пароль для подключения к удаленному серверу (не скрыт). Значение по умолчанию: пустая строка.
  • secure - использовать ли безопасное SSL/TLS-соединение. Обычно также требует указания порта (стандартный безопасный порт - 9440). Сервер должен слушать на <tcp_port_secure>9440</tcp_port_secure> и быть настроен с правильными сертификатами.
  • compression - использовать сжатие данных. Значение по умолчанию: true.

При указании реплик одна из доступных реплик будет выбрана для каждого из шардов при чтении. Вы можете настроить алгоритм балансировки нагрузки (предпочтение, какую реплику использовать) – см. настройку load_balancing. Если соединение с сервером не устанавливается, будет предпринята попытка подключения с коротким тайм-аутом. Если подключение не удалось, будет выбрана следующая реплика, и так далее для всех реплик. Если попытка подключения не удалась для всех реплик, попытка будет повторена таким же образом несколько раз. Это работает в интересах устойчивости, но не обеспечивает полной отказоустойчивости: удаленный сервер может принять соединение, но может не работать или работать плохо.

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

Вы можете указать столько кластеров, сколько хотите, в конфигурации.

Чтобы просмотреть свои кластеры, используйте таблицу system.clusters.

Движок Distributed позволяет работать с кластером как с локальным сервером. Однако конфигурация кластера не может быть указана динамически, она должна быть настроена в файле конфигурации сервера. Обычно все серверы в кластере будут иметь одинаковую конфигурацию кластера (хотя это и не обязательно). Кластеры из файла конфигурации обновляются на лету, без перезапуска сервера.

Если вам нужно отправлять запрос к неизвестному набору шардов и реплик каждый раз, вам не нужно создавать таблицу Distributed - используйте вместо этого функцию таблицы remote. См. раздел Функции таблиц.

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

Существует два метода записи данных в кластер:

Первый способ заключается в том, чтобы определить, на какие серверы записывать какие данные и выполнять запись непосредственно на каждом шарде. Другими словами, выполнять прямые INSERT операторы на удаленных таблицах в кластере, к которому ссылается таблица Distributed. Это наиболее гибкое решение, так как вы можете использовать любую схему шардирования, даже такую, которая не тривиальна из-за требований предметной области. Это также самое оптимальное решение, так как данные могут записываться на разные шардов совершенно независимо.

Второй способ состоит в выполнении операторов INSERT на таблице Distributed. В этом случае таблица будет самостоятельно распределять вставленные данные между серверами. Для записи в таблицу Distributed необходимо, чтобы параметр sharding_key был настроен (если шардов только один).

Каждый шард может иметь <weight>, заданный в файле конфигурации. По умолчанию вес составляет 1. Данные распределяются по шардом в количестве, пропорциональном весу шарда. Все веса шардов суммируются, затем вес каждого шарда делится на общую сумму для определения доли каждого шарда. Например, если есть два шардов, и первый имеет вес 1, а второй 2, то первый получит одну треть (1 / 3) вставленных строк, а второй получит две трети (2 / 3).

Каждый шард может иметь параметр internal_replication, заданный в файле конфигурации. Если этот параметр установлен в true, операция записи выбирает первую здоровую реплику и записывает данные в нее. Используйте это, если таблицы, которые лежат в основе таблицы Distributed, являются реплицированными таблицами (например, любые движки таблиц Replicated*MergeTree). Одна из реплик таблицы получит запись, и она будет автоматически реплицироваться на остальные реплики.

Если internal_replication установлен в false (по умолчанию), данные записываются на все реплики. В этом случае таблица Distributed реплицирует данные сама. Это хуже, чем использование реплицированных таблиц, поскольку согласованность реплик не проверяется, и со временем они будут содержать слегка разные данные.

Чтобы выбрать шард, которому будет отправлена строка данных, выражение шардирования анализируется, и его остаток берется от деления на общий вес шардов. Строка отправляется на шард, который соответствует полуинтервалу остатков от prev_weights до prev_weights + weight, где prev_weights – это общий вес шардов с наименьшим значением, а weight – это вес этого шард. Например, если есть два шарда, и первый имеет вес 9, а второй имеет вес 10, строка будет отправлена на первый шард для остатков из диапазона [0, 9), а на второй для остатков из диапазона [9, 19).

Выражение шардирования может быть любым выражением из констант и столбцов таблицы, которое возвращает целое число. Например, вы можете использовать выражение rand() для случайного распределения данных или UserID для распределения по остатку от деления ID пользователя (в этом случае данные одного пользователя будут находиться на одном шарде, что упрощает выполнение IN и JOIN по пользователям). Если один из столбцов распределен недостаточно равномерно, вы можете обернуть его в хеш-функцию, например, intHash64(UserID).

Простое деление по остаткам - это ограниченное решение для шардирования и не всегда подходит. Оно работает для средних и больших объемов данных (десятки серверов), но не для очень больших объемов данных (сотни серверов и более). В последнем случае используйте схему шардирования, соответствующую предметной области, а не вводите записи в таблицы Distributed.

Вы должны обратить внимание на схему шардирования в следующих случаях:

  • Используются запросы, требующие соединения данных (IN или JOIN) по определенному ключу. Если данные шардированы по этому ключу, вы можете использовать локальные IN или JOIN вместо GLOBAL IN или GLOBAL JOIN, что намного эффективнее.
  • Используется большое количество серверов (сотни и более) с большим количеством небольших запросов, например, запросы на данные отдельных клиентов (например, веб-сайтов, рекламодателей или партнеров). С точки зрения эффективности имеет смысл размещать данные для одного клиента на одном шарде, чтобы небольшие запросы не затрагивали весь кластер. Также можно настроить двухуровневое шардирование: разделить весь кластер на "слои", где слой может состоять из нескольких шардов. Данные для одного клиента размещаются на одном слое, но шардом можно добавлять к слою по мере необходимости, а данные распределяются случайным образом внутри них. Для каждого слоя создаются таблицы 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, вставленные данные могут быть утеряны. Если в директории таблицы обнаружена поврежденная часть данных, она переносится в подсистему broken и больше не используется.

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

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

Когда опция max_parallel_replicas включена, обработка запросов распараллеливается по всем репликам в пределах одного шарда. Для получения дополнительной информации смотрите раздел max_parallel_replicas.

Чтобы узнать больше о том, как обрабатываются распределенные in и global in запросы, обратитесь к этой документации.

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

_shard_num​

_shard_num — Содержит значение shard_num из таблицы system.clusters. Тип: UInt32.

примечание

Поскольку функции таблиц remote и cluster внутренне создают временную таблицу Distributed, _shard_num доступен и там.

Смотрите также