Шардинг в Manticore Search: автоматическое распределение и репликация / Хабр

На старте поисковая система часто устроена просто: одна таблица на одном сервере. Это работает, пока не случится одно из двух. Либо отдельный запрос перестаёт задействовать весь CPU, за который вы заплатили, либо одного сервера перестаёт хватать — по объёму, по пропускной способности или просто потому, что сервер может выйти из строя, и данные на нём будут потеряны.

Автоматический шардинг, встроенный в Manticore Search и доступный начиная с релиза 27.1.5 , решает обе проблемы, разбивая таблицу на несколько физических фрагментов меньшего размера (шардов), по которым можно выполнять поиск параллельно и которые можно размещать на разных узлах:

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

Второй сценарий — главная причина, ради которой чаще всего включают шардинг: высокая доступность. Вы задаёте, сколько шардов хотите и сколько копий каждого должно существовать, а Manticore Search берёт на себя размещение, репликацию и ребалансировку. Вам не нужно писать собственные сценарии переключения при отказах.

Дальше разберём оба сценария, внутреннее устройство без лишнего погружения в детали, команды, которые вы будете запускать, и текущие ограничения.

Краткий глоссарий

Дальше понадобятся такие термины:

Термин	Значение
Шард	Один физический фрагмент таблицы — настоящая таблица, которую Manticore Search создаёт и обслуживает за вас. У таблицы с `shards='4'` их четыре.
Реплика	Копия шарда на другом узле. Благодаря репликам данные сохраняются при отказе узла.
Фактор репликации (RF)	Сколько узлов хранят копию каждого шарда. `rf='2'` означает, что каждый шард существует на двух узлах.
Распределённая таблица	Таблица, с которой вы работаете. У неё заданное вами имя, и она прозрачно рассылает запросы по всем шардам.
Кластер	Кластер репликации Manticore Search — группа узлов, между которыми реплицируются данные.
Мастер	Узел, который сейчас координирует операции шардинга (размещение, ребалансировку). Выбирается автоматически.
Ребалансировка	Автоматический процесс, который перемещает или копирует шарды при изменении набора узлов.

Как создать шардированную таблицу

Шардинг в Manticore Search полностью задаётся двумя простыми опциями CREATE TABLE:

CREATE TABLE products (id bigint, title text, price float) shards='4' rf='2'

shards='N' — разбить таблицу на N физических фрагментов.
rf='M' — хранить M копий каждого фрагмента в кластере (фактор репликации).

Обычно достаточно выполнить один такой CREATE TABLE. Нет отдельного шага «сделать это распределённым», нет ручных списков agent=, как было раньше при ручном шардинге, и нет создания таблицы на каждом узле. Manticore Search создаёт физические шарды, размещает их, настраивает репликацию и создаёт распределённую таблицу с именем products на каждом узле, так что к ней можно одинаково обращаться с любого узла кластера.

Сценарий A: шардинг на одном узле

Начнём со случая, когда у вас один сервер — возможно, большой, с множеством ядер — и пока нет кластера. В таком режиме шардинг нужен не для надёжности хранения: он помогает лучше использовать ресурсы этой машины. Если вся запись идёт в одну real-time таблицу, одновременные INSERT-запросы конкурируют за одни и те же внутренние блокировки. По мере роста таблицы слияния RAM-чанков становятся тяжелее и могут задерживать загрузку данных. Разбиение этой таблицы на несколько независимых шардов решает обе задачи: запись распределяется по шардам, а каждый фрагмент остаётся небольшим. О высокой доступности здесь речь пока не идёт — для неё нужно больше одного узла — поэтому этот сценарий про производительность, а не отказоустойчивость.

Простейшая форма — без кластера и с rf='1':

CREATE TABLE logs (id bigint, message text, ts timestamp) shards='8' rf='1'

Это создаёт восемь физических шардов на одном узле и распределённую таблицу logs, которая объединяет их. Как это помогает на отдельной машине?

Больше параллелизма при загрузке. Каждый шард — это независимая real-time таблица, поэтому одновременные записи распределяются по ним, а не выстраиваются в очередь на блокировках одной таблицы — именно этот выигрыш напрямую измеряют бенчмарки ниже .
Небольшие фрагменты остаются быстрыми. Real-time таблицы периодически сливают свои внутренние RAM-чанки. У таблицы, разбитой на шарды, чанки каждого шарда меньше, поэтому такие слияния требуют меньше ресурсов и реже замедляют вставки.
Параллелизм запросов (на одной машине эффект обычно небольшой). Распределённая таблица ищет по своим шардам параллельно, используя пул воркеров сервера, так что один запрос может задействовать несколько ядер вместо одного — в пределах searchd.threads и числа физических ядер. Впрочем, на одном узле этот механизм частично дублирует псевдошардинг , и выигрыш обычно небольшой (~5–12%) — см. бенчмарки по чтению ниже .

Если вы раньше пользовались псевдошардингом Manticore Search, цель вам может быть знакома — задействовать все ядра для одного запроса — но механизм другой. Псевдошардинг автоматически распараллеливает одну физическую таблицу во время запроса. Явный шардинг создаёт настоящие шарды под вашим контролем: вы задаёте их количество, они существуют как отдельные таблицы, и — что важно — ту же шардированную таблицу позже можно распределить по узлам, не меняя того, как с ней работает приложение.

Эти два механизма дополняют друг друга, но их эффекты не складываются автоматически. Физический шардинг — распределённая таблица поверх нескольких локальных — уже задействует воркеры, поэтому если вы явно шардировали таблицу, включение pseudo_sharding поверх обычно даёт мало и может даже немного снизить пропускную способность. Проверьте оба варианта с помощью manticore-load : прогоните свою нагрузку с pseudo_sharding и без него, и если поверх явных шардов он ничего не добавляет — отключите его.

На одном узле фактор репликации должен быть 1: ведь узел всего один, поэтому второй копии негде разместиться. В этом и подвох — шардинг на одном узле даёт параллелизм, но не надёжность. Для надёжности нужно больше одного узла.

Сценарий B: шардинг на нескольких узлах и автоматическая репликация

Это основной сценарий для шардинга. Начните с кластера репликации из нескольких узлов (как его создать — см. Настройка репликации ), затем создайте таблицу внутри этого кластера с префиксом cluster: и RF больше 1:

CREATE TABLE mycluster:products (id bigint, title text, price float) shards='4' rf='2'

Вот что Manticore Search делает за вас:

Создаёт четыре шарда.
Размещает их по узлам кластера сбалансированно.
Создаёт вторую копию каждого шарда на другом узле, потому что rf='2'.
Настраивает репликацию между каждым шардом и его репликой.
Создаёт распределённую таблицу products на каждом узле, так что любой узел может обслуживать чтение и принимать запись.

С точки зрения приложения ничего не изменилось — вы всё так же делаете INSERT INTO products … и SELECT … FROM products. При чтении Manticore распределяет запрос по шардам и сливает результаты; операцию записи направляет в нужный шард. Но теперь каждый шард хранится на двух узлах, и именно это важно: любой отдельный узел может выйти из строя, а таблица останется полностью доступной без потери данных.

Фактор репликации масштабируется в зависимости от ваших требований к надёжности и числа узлов:

RF	Копий на шард	Отказы без потери данных	Типичное применение
1	1	нет — при отказе узла теряются его шарды	параллелизм на одном узле, dev/test, данные, которые можно восстановить
2	2	один узел	обычный выбор для продакшена
3+	3 и более	несколько узлов одновременно	критически важные системы, среды с частыми отказами

Ограничение простое: нельзя запросить больше копий, чем у вас узлов. rf='3' требует минимум три узла в кластере. Manticore Search проверяет это при создании таблицы и сообщит, если кластер слишком мал.

-- 6 shards, 3 copies each, across a 3+ node cluster
CREATE TABLE mycluster:events (id bigint, body text) shards='6' rf='3'

Соберём всё вместе: пошаговый пример на нескольких узлах

Допустим, у вас есть кластер репликации из трёх узлов под названием mycluster (если его ещё нет, см. про Настройку репликации командами CREATE CLUSTER и JOIN CLUSTER). Создайте шардированную реплицируемую таблицу с любого узла:

CREATE TABLE mycluster:products (id bigint, title text, price float) shards='4' rf='2'

Manticore Search создаёт четыре шарда, размещает по две копии каждого шарда на трёх узлах и создаёт распределённую таблицу products на каждом узле. Проверьте размещение:

SHOW SHARDING STATUS products;

-- illustrative output (abbreviated columns)
+-------+-------+--------+----+-----------+
| shard | node  | status | rf | rf_status |
+-------+-------+--------+----+-----------+
|     0 | node1 | active |  2 | ok        |
|     0 | node2 | active |  2 | ok        |
|     1 | node2 | active |  2 | ok        |
|     1 | node3 | active |  2 | ok        |
|     2 | node1 | active |  2 | ok        |
|     2 | node3 | active |  2 | ok        |
|     3 | node1 | active |  2 | ok        |
|     3 | node2 | active |  2 | ok        |
+-------+-------+--------+----+-----------+

Каждый шард присутствует на двух разных узлах — это и есть rf=2 — и у каждого rf_status равен ok. (В полном результате есть ещё столбцы table, cluster и replication_cluster.) Теперь с любого узла обращайтесь к ней как к обычной таблице:

INSERT INTO products (id, title, price) VALUES (1, 'Wireless mouse', 19.99);
SELECT * FROM products WHERE MATCH('mouse');

Запись направляется в шард и реплицируется во вторую копию этого шарда; при чтении Manticore обращается ко всем четырём шардам и сливает результаты. Ваше приложение нигде не указывает шард по имени.

Поддержание фактора репликации

Задать rf='2' легко. Сложное в любой распределённой системе — соблюдать это условие со временем, по мере того как машины выходят из строя и возвращаются, а вы добавляете мощности. Но вам больше беспокоиться об этом не нужно. Эту работу Manticore Search автоматизирует.

Работает это так: кластер выбирает мастер-узел, который запускает цикл координации. Он следит за топологией кластера — какие узлы живы — и реагирует на изменения:

Узел выходит из строя

У его шардов теперь меньше копий, чем требует RF. Мастер обнаруживает пропавший узел и пытается создать недостающие реплики заново. Если после отказа в кластере осталось хотя бы rf активных узлов, он размещает новые копии на активных узлах, где этих реплик ещё нет, и восстанавливает фактор репликации. Запросы продолжают работать, пока хотя бы одна копия каждого шарда остаётся доступной.

Продолжая пример выше — если node3 падает, SHOW SHARDING STATUS products показывает затронутые шарды как degraded (одна копия недоступна, одна ещё жива):

-- illustrative: node3 is down
+-------+-------+----------+----+-----------+
| shard | node  | status   | rf | rf_status |
+-------+-------+----------+----+-----------+
|     1 | node2 | active   |  2 | degraded  |
|     1 | node3 | inactive |  2 | degraded  |
|     2 | node1 | active   |  2 | degraded  |
|     2 | node3 | inactive |  2 | degraded  |
|   ... | ...   | ...      |    | ...       |
+-------+-------+----------+----+-----------+

Остаются два активных узла (node1, node2) и rf=2, поэтому мастер создаёт недостающие копии шардов 1 и 2 на том активном узле, где их ещё нет. Пока новая копия строится, она отображается как pending; как только репликация догоняет, она становится active, а rf_status возвращается к ok.

Важная оговорка: Manticore Search может восстановить RF только если есть куда поместить новую копию. Если активных узлов стало меньше, чем rf, полностью соблюсти RF сейчас невозможно: затронутые шарды остаются degraded с уцелевшей копией, пока узел не вернётся или вы не добавите новый. Manticore Search не станет создавать вторую копию того же шарда на том же узле и не будет молча делать вид, что RF соблюдён. Если живых копий шарда не осталось совсем, статус становится broken; этот случай описан ниже. При rf='1' шарды отказавшего узла просто исчезают — второй копии никогда и не было.

Узел присоединяется

Новые ресурсы нужно срочно задействовать. Мастер выполняет ребалансировку, чтобы новый узел взял на себя свою долю нагрузки. Как именно — зависит от RF:

RF = 1: шарды нужно перемещать (копия всего одна, поэтому её нельзя просто продублировать). Manticore Search перемещает их безопасно, используя временный внутренний кластер: сначала копирует данные на новый узел и только потом удаляет их со старого, так что у шарда всё время есть доступная копия.
RF ≥ 2: шарды реплицируются на новый узел через уже имеющуюся репликацию кластера, затем распределение перебалансируется. Никакого рискованного перемещения данных, потому что вторая копия всегда есть.

Все копии шарда недоступны

Если все узлы, на которых хранится этот шард, выходят из строя одновременно, rf_status этого шарда становится broken — не остаётся ни одной живой копии, которая могла бы обслуживать запросы или служить источником репликации. Остальная таблица продолжает работать; этот шард восстанавливается, когда один из его узлов возвращается. RF снижает вероятность таких ситуаций: при rf='2' нужны два одновременных отказа именно тех узлов, на которых хранится шард, при rf='3' — три.

Всё это происходит через внутреннюю упорядоченную очередь операций с поддержкой отката, так что операция ребалансировки либо завершается, либо корректно откатывается — даже если сам мастер-узел выходит из строя посреди операции, следующий мастер убирает незавершённую работу. Что это значит для вас как для оператора: вы задаёте RF один раз, а кластер следит, чтобы он соблюдался.

Как это устроено под капотом (кратко)

Это не обязательно знать, чтобы пользоваться шардингом, но понимать, что происходит, полезно.

Физические шарды — это настоящие таблицы. За таблицей с четырьмя шардами стоят четыре настоящие таблицы, которые Manticore Search создаёт и обслуживает за вас. Обычно вы напрямую к ним не обращаетесь.
Приложение работает с распределённой таблицей . Manticore Search создаёт такую таблицу с именем products на каждом узле. В её внутренней схеме локальные шарды указаны напрямую, а шарды на других узлах подключены через agent. Именно это позволяет SELECT … FROM products прозрачно обращаться ко всем шардам.
Состояние координации хранится в кластере. Manticore Search ведёт собственные внутренние метаданные — размещение шардов, состояние координации и очередь ожидающих операций — поэтому всегда знает, где что хранится и какая работа ещё не выполнена. В конфигурации с несколькими узлами это состояние реплицируется по кластеру, так что у каждого узла одно и то же представление состояния.
Изменениями управляет мастер. Размещение, настройку репликации и ребалансировку вычисляет мастер и помещает в очередь как упорядоченные команды с инструкциями отката, после чего они выполняются на узлах.
Репликация переиспользует кластеризацию Manticore Search. Тот же проверенный механизм репликации, который Manticore Search уже использует для кластеров, держит реплики шардов синхронизированными.

Архитектурно:

            CREATE TABLE ... shards='4' rf='2'
                          │
                          ▼
                  ┌────────────────┐
                  │ выбранный      │  вычисляет размещение,
                  │ мастер         │  ставит операции в очередь
                  └────────┬───────┘
                          │
        ┌─────────────────┼─────────────────┐
        ▼                 ▼                 ▼
    ┌──────────┐     ┌──────────┐     ┌──────────┐
    │  node1   │     │  node2   │     │  node3   │
    │ s0 s2 s3 │◄───►│ s0 s1 s3 │◄───►│  s1 s2   │   (каждый шард на 2 узлах = RF=2)
    └──────────┘     └──────────┘     └──────────┘
    распределённая таблица "products" создана на каждом узле
    (то же размещение, что в SHOW SHARDING STATUS выше)

Эксплуатация шардированной таблицы

Тут всё просто: работает всё, чего вы ожидаете от обычной таблицы, плюс есть пара команд, специфичных для шардинга, чтобы видеть его состояние.

Посмотреть схему. DESC и SHOW CREATE TABLE работают с логической таблицей; Manticore Search получает нужную информацию через нижележащие шарды:

DESC products;
SHOW CREATE TABLE products;

Узнать, где хранится каждый шард и соблюдается ли RF. Это команда, за которой вы будете следить во время отказов и ребалансировки:

SHOW SHARDING STATUS products;

+-------+-------+--------+----+-----------+
| shard | node  | status | rf | rf_status |
+-------+-------+--------+----+-----------+
|     0 | node1 | active |  2 | ok        |
|     0 | node2 | active |  2 | ok        |
|   ... | ...   | ...    |    | ...       |
+-------+-------+--------+----+-----------+

Она выдаёт по одной строке на копию шарда со следующими столбцами:

Столбец	Значение
`table`	имя таблицы
`shard`	номер шарда
`node`	узел, на котором хранится эта копия
`status`	`active`, `inactive` (узел недоступен) или `pending` (создаётся)
`cluster`	кластер репликации, которому принадлежит таблица
`replication_cluster`	внутренний кластер, который держит копии этого шарда синхронизированными
`rf`	сколько копий сейчас у этого шарда
`rf_status`	`ok` (RF соблюдён), `degraded` (часть копий недоступна, но хотя бы одна жива) или `broken` (живых копий нет)

rf_status — это быстрый индикатор состояния: ok во всех строках означает, что кластер обеспечивает запрошенный вами фактор репликации; degraded — работает, но уязвим; broken — шард недоступен.

Найти координатора:

SHOW SHARDING MASTER;

Корректно удалить. Удаление шардированной таблицы работает точно так же, как удаление обычной таблицы — DROP TABLE удаляет таблицу и все её шарды по всему кластеру:

DROP TABLE products;

Масштабировать через изменение кластера. Поскольку ребалансировка автоматическая, для горизонтального масштабирования шардированной таблицы вы добавляете узлы в кластер (см. Добавление нового узла ). Мастер замечает новый узел и размещает на нём часть шардов или их реплик без каких-либо действий с самой таблицей.

Выбор числа шардов и фактора репликации

Несколько практических правил:

Шарды ради ускорения записи: обычно достаточно небольшого числа шардов — заметно меньше числа ваших ядер — в бенчмарках ниже машина с 16 ядрами / 32 потоками показала пик на 4–8 шардах, а к 32 шардам была медленнее, чем вообще без шардинга. Начните с малого (4–8), измерьте и добавляйте больше, только если ваши собственные цифры это подтверждают. Большее число шардов помогает редко и добавляет накладные расходы на каждый шард.
Шарды ради распределения: при нескольких узлах нужно столько шардов, чтобы они ровно делились по узлам и оставляли запас для роста; хороший вариант по умолчанию — число, кратное количеству узлов. Не переусердствуйте: каждый шард — настоящая таблица со своими накладными расходами. (Manticore Search ограничивает число шардов значением 3000.)
RF ради надёжности: rf='2' — стандартный выбор для продакшена: он выдерживает отказ любого одного узла ценой двойного объёма хранения. Используйте rf='3' только когда вам действительно нужно пережить одновременные отказы или есть строгие требования к доступности, и помните, что это стоит тройного объёма хранения и большего трафика репликации.
RF=1 — только для производительности или одноразовых данных. В этом режиме нет отказоустойчивости. Используйте его на одном узле ради распараллеливания записи или в кластере только тогда, когда у вас есть внешний способ восстановить потерянные данные.

Бенчмарки: действительно ли шардинг ускоряет вставки?

Любой новой функциональностью стоит пользоваться, только если это себя оправдывает, поэтому мы измерили. Вопрос, на который мы хотели честно ответить: становится ли загрузка данных быстрее при одной и той же нагрузке — и если да, то в каком случае? Методика простая: каждый вариант с шардингом сравнивался с одним и тем же базовым уровнем — обычной таблицей без шардинга.

Если коротко: на машине с 16 ядрами и 32 одновременными пишущими потоками шардинг поднял пропускную способность вставок примерно в 1,5× в лучшем случае — с ~163k до ~253k док/с — но только пока число шардов оставалось небольшим. Лучший результат получился на 4–8 шардах; к 32 шардам пропускная способность упала ниже базового уровня без шардинга. Бинарный лог снижал производительность записи примерно на 25%, а репликация rf=2 между двумя реальными машинами — ещё примерно на 30%: это ожидаемая, но заметная плата за надёжность.

Стенд. Выделенный сервер без другой заметной нагрузки на CPU — AMD Ryzen 9 5950X (16 ядер / 32 потока), 128 ГБ RAM. Всё работало внутри одного Docker-контейнера со свежей dev-сборкой с функцией шардинга. Manticore Search работал с настройками по умолчанию — без тюнинга производительности: задавались только порты слушателей, каталог данных и путь к бинарному логу; пул потоков, лимиты памяти RT и поведение бинарного лога остались по умолчанию. Нагрузку давал manticore-load . В каждом прогоне вставляются одни и те же документы — (id bigint, name text, type int), где name — это 10–100 случайных слов — пачками по 1000 в real-time таблицу. Между прогонами меняются только число шардов, фактор репликации и бинарный лог; базовый уровень «без шардинга» — это обычная RT-таблица. Полную серию прогонов по числу шардов на одном узле мы запускаем дважды — один раз с включённым бинарным логом (по умолчанию), один раз с выключенным — вставляя по 20 000 000 документов за прогон с 32 одновременными пишущими потоками.

# the exact shape of every insert run (shards/rf vary)
manticore-load --batch-size=1000 --threads=32 --total=20000000 \
  --init="create table test(id bigint, name text, type int) shards='8' rf='1'" \
  --load="insert into test(id,name,type) values(<increment>,'<text/10/100>',<int/1/100>)"

Один узел: пропускная способность в зависимости от числа шардов

Пропускная способность вставок в зависимости от числа шардов, бинарный лог вкл/выкл, на одном узле 16 ядер / 32 потока

Вставок в секунду, 20M документов, 32 пишущих потока — оба режима бинарного лога:

Шарды	бинарный лог вкл (по умолчанию)	бинарный лог выкл
нет (базовый уровень)	162,920	218,079
2	191,976	246,288
4	252,807	290,665
8	251,008	265,015
16	175,848	182,288
32	108,006	111,381

На графике полный прогон показан дважды — бинарный лог вкл (синий, по умолчанию) и выкл (оранжевый). Видны три вещи:

Параллельные вставки действительно ускоряются. При 32 пишущих потоках разбиение таблицы поднимает пропускную способность на ~1,5× на пике — с 163k до 253k док/с в режиме по умолчанию (с бинарным логом). Каждый шард — независимая real-time таблица, поэтому распределение записи по нескольким шардам резко снижает конкуренцию за блокировки, с которой сталкивается одна RT-таблица под параллельной нагрузкой, и позволяет вставкам задействовать больше ядер.
Максимум достигается уже на 4–8 шардах. Затем выигрыш быстро падает. К 16 шардам он едва выше базового уровня, а на 32 шардах пропускная способность ниже, чем у нешардированной таблицы (0,66×) — дальше накладные расходы на шарды и координацию перевешивают дополнительный параллелизм. Больше шардов — точно не лучше. Обе линии имеют одинаковую форму и дают максимум в одном и том же диапазоне — бинарный лог этого не меняет.
Надёжность дороже всего там, где пропускная способность максимальна. Отключение бинарного лога (оранжевый) поднимает всю кривую, но разрыв шире всего в области высокой пропускной способности — 253k → 291k на 4 шардах — и почти исчезает при большом числе шардов (108k против 111k на 32 шардах), где узкое место — накладные расходы на координацию, а не надёжность. Ниже отдельно покажем влияние бинарного лога.

Одна важная оговорка: ускорение появляется именно при параллельной записи. Один строго последовательный пишущий поток не может воспользоваться параллельными шардами и не увидит ускорения (и получит немного накладных расходов распределённого слоя). Шардинг окупается, когда пишут сразу много клиентов или потребителей — а именно так и работают реальные конвейеры загрузки данных.

Цена надёжности: бинарный лог

Бинарный лог Manticore Search делает вставки устойчивыми к сбоям (он может воспроизвести несброшенные транзакции после нечистого завершения). Он включён по умолчанию. Его цену уже видно на графике выше — оранжевая линия (без бинарного лога) идёт выше синей. Посмотрим на неё отдельно на обычной нешардированной таблице, меняя только binlog_path:

Пропускная способность вставок с бинарным логом вкл/выкл (20M документов)

Отключение бинарного лога подняло базовую пропускную способность с 162k до 218k док/с — то есть устойчивость вставок к сбоям снижает производительность примерно на 25%. Это плата за то, чтобы не терять несброшенные записи при жёстком сбое; в продакшене оставляйте его включённым. Отключать бинарный лог стоит только если данные можно восстановить из источника и на массовых загрузках вам важнее дополнительная скорость.

Цена репликации для записи

Надёжность между машинами тоже не бесплатна — и этот тест честен только на реальных отдельных машинах. Поэтому, в отличие от одноузловых бенчмарков выше, два теста репликации мы запускали на двух разных физических машинах: это были две облачные VM по 4 ядра / 7 ГБ, объединённые в кластер Manticore Search. Так rf=1 и rf=2 никогда не конкурируют за одни и те же ядра или память — честное сравнение, которое одна машина просто не может дать.

rf=1 хранит одну копию на одной машине; rf=2 хранит полную копию на обеих, поэтому каждая вставка синхронно реплицируется по сети до подтверждения. Та же нагрузка в 1M документов, 4 пишущих потока:

Пропускная способность вставок при rf=1 и rf=2 на двух отдельных машинах

Репликация снизила пропускную способность вставок примерно на 30% (112k → 78k док/с) — это плата за копирование каждой записи на вторую машину до подтверждения. Вот компромисс rf в одном числе: пишете чуть медленнее, зато данные сохраняются при потере целой машины.

Ускоряют ли шардинг и репликация чтение?

Чтение легко измерить неправильно, поэтому сначала о методике. Тривиальный запрос — получить 20 строк без ранжирования — выполняется за время значительно меньше миллисекунды, так что почти всё его время уходит на фиксированные накладные расходы; распределённая таблица, которая обращается к агенту, выглядит тогда в три раза медленнее (≈7700 против 2400 запр/с здесь) только потому, что пинг занимает гораздо больше времени, чем почти нулевая полезная работа. Такой запрос непоказателен. Реалистичный полнотекстовый запрос (несколько слов для поиска и ранжирования, ~10–30 мс) даёт более честную картину: накладные расходы на распределение снижаются до нескольких процентов, потому что теперь основное время уходит на реальную работу, а не на фиксированные накладные расходы. Все цифры ниже используют реалистичные запросы на кластере из двух узлов, читая с одного входного узла.

# realistic read — what the numbers below use: full-text match-and-rank, ~10–30 ms each
manticore-load --threads=4 --total=5000 \
  --load="select id from <table> where match('<text/1/5> <text/1/5>')"

# trivial read — sub-ms, all fixed overhead, exaggerates distributed cost ~3x
manticore-load --threads=4 --total=20000 \
  --load="select id from <table> limit 20 option ranker=none"

<table> — это одно из: обычная RT-таблица (без шардинга), таблица type='distributed' поверх локальных шардов (шардинг на одном узле — local='s0' local='s1' …) или шардированная таблица с shards='4' rf='1'/rf='2'. Каждое чтение направляется на один узел — мы не запускаем отдельные клиентские запросы к обоим узлам.

Пропускная способность чтения: шардинг на двух узлах

Шардинг по узлам ускоряет чтение. Таблица из 4 шардов с rf=1, распределённая по обеим машинам, выдаёт ~516 запр/с против ~315 у одного нешардированного узла — примерно 1,6× — потому что каждый запрос выполняется сразу на ядрах обоих узлов.
Репликация сохраняет скорость чтения. rf=2 (каждый шард на обоих узлах) выдаёт ~410 запр/с — чуть ниже rf=1. Причина не в том, что нагрузка привязана к одному узлу (она распределяется по репликам); дело в том, что при rf=2 к каждому шарду обращаются через путь агента/зеркала — даже к тем копиям, что оказались локальными — и этот путь добавляет те самые несколько процентов накладных расходов, отмеченные выше, тогда как rf=1 читает свои локальные шарды внутри процесса. Так или иначе чтение остаётся заметно выше одного узла, и теперь данные сохраняются при потере машины.
На одной машине выигрыш при чтении от шардинга невелик. Разбиение таблицы на 2–4 шарда на одном узле добавляет лишь ~5–12% (параллелизм запросов), и только когда есть свободные ядра — тяжёлые запросы выигрывают больше (≈12% против ≈3% для лёгких), но на полностью загруженной машине эффект нулевой. Реальное масштабирование чтения даёт добавление машин, а не шардов на одной машине.

Замечание о методике: тесты чтения и репликации запускались на отдельной паре облачных VM — по 4 vCPU / 7 ГБ RAM, две разные физические машины — объединённых в кластер из двух узлов, та же свежая dev-сборка, настройки по умолчанию, 1,5M документов. Это другой (и гораздо менее мощный) стенд, чем сервер с 16 ядрами / 32 потоками, использованный для прогонов вставок выше, поэтому делайте выводы внутри каждого теста, а не между ними. Малое число ядер — ещё и причина, почему выигрыш чтения от шардинга на одном узле здесь скромный: для распараллеливания запроса свободных ядер мало.

Итог

На одной машине шардинг даёт ~1,5× ускорение параллельных вставок (здесь: 163k → 253k док/с на 20M документов).
Лучший результат здесь — на 4–8 шардах для CPU с 16 ядрами / 32 потоками. Слишком много шардов вредит: на 32 шардах пропускная способность упала ниже базового уровня без шардинга. Подбирайте число шардов под ядра и число одновременных пишущих потоков, а не под большое круглое число.
Бинарный лог снижает пропускную способность записи примерно на 25% (и разрыв сжимается почти до нуля при слишком большом числе шардов), а репликация rf=2 — примерно на 30% для записи (измерено на двух отдельных машинах) — это плата за устойчивость к сбоям и за сохранение данных при отказе узла соответственно.
С реалистичными полнотекстовыми запросами шардинг по кластеру из двух узлов даёт примерно 1,6× пропускной способности чтения относительно одного узла, а rf=2 сохраняет скорость чтения; данные при этом остаются доступными при потере узла. Тривиальные запросы по id преувеличивают накладные расходы распределения примерно в 3 раза — всегда измеряйте чтение реалистичными запросами.
Абсолютные числа зависят от железа и формы документов; от конфигурации к конфигурации переносится характер этих кривых — поэтому измерьте свою нагрузку, прежде чем фиксировать число шардов.

Ограничения и что важно знать

Есть острые углы, о которых стоит знать заранее:

rf обязателен. Шардированный CREATE TABLE должен указывать rf=. Если его опустить, CREATE TABLE завершится ошибкой.
rf больше 1 требует кластера. Нельзя создать шардированную таблицу с несколькими копиями на отдельном узле — копии негде разместить. Таблицы с несколькими копиями должны использовать форму cluster:name.
Локальные шардированные таблицы нельзя создавать на узле, который уже в кластере. Если узел принадлежит кластеру репликации, создавайте таблицу в этом кластере (CREATE TABLE cluster:name …), а не как локальную, иначе метаданные шардинга не будут отслеживаться корректно. Manticore Search обнаруживает это и сообщает вам.
Максимум 3000 шардов на таблицу.
RF=1 означает отсутствие отказоустойчивости. Шарды потерянного узла пропадают. Это свойство режима, а не баг — это компромисс, на который вы соглашаетесь при rf='1'.
Нужно минимум RF узлов. rf='M' требует кластера минимум из M узлов; иначе создание не удастся.
Создание синхронное, но ограничено тайм-аутом. CREATE TABLE ждёт завершения распределения (по умолчанию 30 с). Для очень большого числа шардов увеличьте тайм-аут через timeout='N' (в секундах), например shards='3000' rf='3' timeout='60'.

Что в итоге

Шардинг в Manticore Search покрывает широкий диапазон сценариев с намеренно простым интерфейсом. В варианте shards='N' rf='1' на одной машине шардинг распределяет одновременные записи по независимым фрагментам и сохраняет каждый фрагмент небольшим. А shards='N' rf='M' внутри кластера даёт вам распределённую реплицируемую таблицу, которая остаётся доступной при отказах узлов и сама перебалансируется при изменении кластера — без единой строчки логики переключения при отказах с вашей стороны. Одна и та же схема таблицы работает хоть на одном узле, хоть на многих, и приложение всё это время обращается к ней одинаково. На практике это значит, что вы можете, например, начать с ускорения записи на одной машине, а затем перейти к отказоустойчивому кластеру без изменений в приложении.

Чтобы глубже разобраться в компонентах, на которых построен шардинг, советуем прочитать документацию:

Остались вопросы или есть идеи по нагрузке, под которой нам стоит протестировать шардинг? Напишите нам .