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

Движок Распределенной Таблицы

осторожно

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

Таблицы с движком 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 - (по желанию) имя политики, которое будет использоваться для хранения временных файлов для фоновой отправки.

См. Также

Настройки Распределенной Таблицы​

fsync_after_insert​

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

fsync_directories​

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

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

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

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

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

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

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

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

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

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

Чтобы выбрать шард, на который будет отправлена строка данных, анализируется выражение шардирования, и берется его остаток от деления на общий вес шардов. Строка отправляется в шард, который соответствует полуинтервалу остатков от 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_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 внутренне создают временные распределенные таблицы, _shard_num доступен и там.

См. Также