Архитектура | Tarantool
Шардинг / Архитектура
Шардинг / Архитектура



Consider a distributed Tarantool cluster that consists of subclusters called shards, each storing some part of data. Each shard, in its turn, constitutes a replica set consisting of several replicas, one of which serves as a master node that processes all read and write requests.

The whole dataset is logically partitioned into a predefined number of virtual buckets (further just buckets), each assigned a unique number ranging from 1 to N, where N is the total number of buckets. The number of buckets is specifically chosen to be several orders of magnitude larger than the potential number of cluster nodes, even given future cluster scaling. For example, with M projected nodes the dataset may be split into 100 * M or even 1,000 * M buckets. Care should be taken when picking the number of buckets: if too large, it may require extra memory for storing the routing information; if too small, it may decrease the granularity of rebalancing.

Each shard stores a unique subset of buckets, which means that a bucket cannot belong to several shards at once, as illustrated below:


This shard-to-bucket mapping is stored in a table in one of Tarantool’s system spaces, with each shard holding only a specific part of the mapping that covers those buckets that were assigned to this shard.

Apart from the mapping table, the bucket id is also stored in a special field of every tuple of every table participating in sharding.

Once a shard receives any request (except for SELECT) from an application, this shard checks the bucket id specified in the request against the table of bucket ids that belong to a given node. If the specified bucket id is invalid, the request gets terminated with the following error: “wrong bucket”. Otherwise the request is executed, and all the data created in the process is assigned the bucket id specified in the request. Note that the request should only modify the data that has the same bucket id as the request itself.

Storing bucket ids both in the data itself and the mapping table ensures data consistency regardless of the application logic and makes rebalancing transparent for the application. Storing the mapping table in a system space ensures sharding is performed consistently in case of a failover, as all the replicas in a shard share a common table state.

The sharded dataset is partitioned into a large number of abstract nodes called virtual buckets (further just buckets).

Секционирование набора данных происходит с помощью сегментного ключа (или идентификатора сегмента (bucket id) в терминах Tarantool’а). Идентификатор сегмента – это число от 1 до N, где N – это общее количество сегментов.


В каждом наборе реплик есть уникальное подмножество сегментов. Один сегмент не может относиться к нескольким наборам реплик одновременно.

Общее количество сегментов определяет администратор, который настраивает первоначальную конфигурацию кластера.

Every space you plan to shard must have a numeric field containing bucket id-s. This field must comply with the following requirements:

  • The field’s data type can be: unsigned, number or integer.
  • The field must be not nullable.
  • The field must be indexed by the shard_index. The default name for this index is bucket_id.

See the configuration example.

A sharded cluster in Tarantool consists of:

  • storages,
  • routers,
  • and a rebalancer.

Storage is a node storing a subset of the dataset. Multiple replicated (for redundancy) storages comprise a replica set (also called shard).

Each storage in a replica set has a role, master or replica. A master processes read and write requests. A replica processes read requests but cannot process write requests.


Router is a standalone software component that routes read and write requests from the client application to shards.

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

  • номер и местоположение шардов,
  • процесс балансировки данных,
  • наличие отказа и восстановление после отказа реплики.

A router can also calculate a bucket id on its own provided that the application clearly defines rules for calculating a bucket id based on the request data. To do it, a router needs to be aware of the data schema.

У роутера нет постоянного статуса, он не хранит топологию кластера и не выполняет балансировку данных. Роутер – это автономный компонент ПО, который может работать на уровне хранилища или на уровне приложения в зависимости от функций приложения.

A router maintains a constant pool of connections to all the storages that is created at startup. Creating it this way helps avoid configuration errors. Once a pool is created, a router caches the current state of the _vbucket table to speed up the routing. In case a bucket id is moved to another storage as a result of data rebalancing, or one of the shards fails over to a replica, a router updates the routing table in a way that’s transparent for the application.

Sharding is not integrated into any centralized configuration storage system. It is assumed that the application itself handles all the interactions with such systems and passes sharding parameters. That said, the configuration can be changed dynamically - for example, when adding or deleting one or several shards:

  1. To add a new shard to the cluster, a system administrator first changes the configuration of all the routers and then the configuration of all the storages.
  2. The new shard becomes available to the storage layer for rebalancing.
  3. As a result of rebalancing, one of the vbuckets is moved to the new shard.
  4. When trying to access the vbucket, a router receives a special error code that specifies the new vbucket location.

CRUD operations can be:

  • executed in a stored procedure inside a storage, or
  • initialized by the application.

In any case, the application must include the operation bucket id in a request. When executing an INSERT request, the operation bucket id is stored in a newly created tuple. In other cases, it is checked if the specified operation bucket id matches the bucket id of a tuple being modified.

Since a storage is not aware of the mapping between a bucket id and a primary key, all the SELECT requests executed in stored procedures inside a storage are only executed locally. Those SELECT requests that were initialized by the application are forwarded to a router. Then, if the application has passed a bucket id, a router uses it for shard calculation.

There are several ways of calling stored procedures in cluster replica sets. Stored procedures can be called:

  • on a specific vbucket located in a replica set (in this case, it is necessary to differentiate between read and write procedures, as write procedures are not applicable to vbuckets that are being migrated), or
  • without specifying any particular vbucket.

All the routing validity checks performed for sharded DML operations hold true for vbucket-bound stored procedures as well.

Rebalancer is a background rebalancing process that ensures an even distribution of buckets across the shards. During rebalancing, buckets are being migrated among replica sets.

The rebalancer «wakes up» periodically and redistributes data from the most loaded nodes to less loaded nodes. Rebalancing starts if the disbalance threshold of a replica set exceeds a disbalance threshold specified in the configuration.

Предел дисбаланса рассчитывается следующим образом:

|эталонное_число_сегментов - текущее_число_сегментов| / эталонное_число_сегментов * 100

Набор реплик, из которого переносится сегмент, называется исходный (source); а набор реплик, куда переносится сегмент, называется целевой (destination).

Блокировка набора реплик позволяет набору реплик оставаться невидимым для балансировщика. Набор реплик с блокировкой не может ни принимать новые сегменты, ни мигрировать свои собственные.

Во время миграции у сегмента могут быть разные статусы:

  • ACTIVE (активный) – сегмент доступен для запросов чтения и записи.
  • PINNED (закрепленный) – сегмент заблокирован для миграции в другой набор реплик. Во всем остальном закрепленные сегменты аналогичны активным сегментам.
  • SENDING (отправляемый) – в настоящий момент сегмент копируется в целевой набор реплик; запросы на чтение в исходный набор реплик обрабатываются.
  • RECEIVING (принимающий) – происходит наполнение сегмента; все запросы отклоняются.
  • SENT (отправленный) – сегмент был перенесен в целевой набор реплик. Роутер использует статус SENT, чтобы определить новое местонахождение сегмента. Сегмент в статусе SENT переходит в статус мусора GARBAGE автоматически через количество секунд, указанное в BUCKET_SENT_GARBAGE_DELAY, по умолчанию равное 0,5 секунды.
  • GARBAGE (мусор) – произошла миграция сегмента в целевой набор реплик во время балансировки; или же принимающий сегмент был в статусе RECEIVING, но произошла ошибка во время миграции.

Сегменты в статусе мусора GARBAGE удаляются сборщиком мусора.


Миграция происходит следующим образом:

  1. В целевом наборе реплик создается новый сегмент, который получает статус RECEIVING (принимающий), начинается копирование данных, и сегмент отклоняет все запросы.
  2. Отправляемый сегмент в исходном наборе реплик получает статус SENDING и продолжает обрабатывать запросы на чтение.
  3. После копирования данных сегмент в исходном наборе реплик получает статус отправленного (SENT) и перестает принимать запросы.
  4. Сегмент в целевом наборе реплик переходит в активный статус (ACTIVE) и начинает принимать все запросы.


There is a specific error vshard.error.code.TRANSFER_IS_IN_PROGRESS that returns in case a request tries to perform an action not applicable to a bucket which is being relocated. You need to retry the request in this case.

Системный спейс _bucket

Системный спейс _bucket в каждом наборе реплик хранит идентификаторы сегментов данного набора реплик. Спейс содержит следующие поля:

  • bucket – идентификатор сегмента
  • status – state of the bucket
  • destination – UUID целевого набора реплик


- - [1, ACTIVE, abfe2ef6-9d11-4756-b668-7f5bc5108e2a]
  - [2, SENT, 19f83dcb-9a01-45bc-a0cf-b0c5060ff82c]

После миграции сегмента UUID целевого набора реплик вносится в таблицу. Пока сегмент еще находится в исходном наборе реплик, значение UUID целевого набора реплик равно NULL.

Таблица маршрутизации роутера отображает все идентификаторы сегментов с соответствующими наборами реплик. Она обеспечивает консистентность шардинга в случае отказа.

Роутер поддерживает постоянный пул соединений со всеми хранилищами, созданными при запуске, что помогает избежать ошибки конфигурации. После создания пула соединений роутер кэширует текущее состояние таблицы маршрутизации, чтобы ускорить ее. Если произошла миграция сегмента в другое хранилище после балансировки или же отказ, который вызвал переключение шарда на другую реплику, файбер обнаружения (discovery fiber) в роутере обновит таблицу маршрутизации автоматически.

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

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

Сначала все запросы направляются в роутер. Роутер поддерживает только операцию вызова, которая выполняется с помощью функции

result =<идентификатор_сегмента>, <режим>, <имя_функции>, {<список_аргументов>}, {<опции>})

Запросы обрабатываются следующим образом:

  1. Роутер использует идентификатор сегмента для поиска набора реплик с соответствующим сегментом в таблице маршрутизации.

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

  2. После обнаружения сегмента шард проверяет:

    • хранится ли сегмент в системном спейсе _bucket набора реплик;
    • находится ли сегмент в статусе ACTIVE (активный) или PINNED (закрепленный) (если выполняется запрос на чтение, то сегмент может находиться в состоянии отправки SENDING).
  3. Если проверка пройдена, запрос выполняется. В противном случае, выполнение запроса прекращается с ошибкой: “wrong bucket” (несоответствующий сегмент).

Вертикальное масштабирование
Добавление мощности в отдельный сервер: использование более мощного процессора, добавление оперативной памяти, добавление хранилищ и т.д.
Горизонтальное масштабирование
Добавление дополнительных серверов в пул ресурсов, последующее секционирование и распределение набора данных по серверам.
Архитектура базы данных, которая допускает секционирование набора данных по сегментному ключу и распределение набора данных по нескольким серверам. Шардинг представляет собой частный случай горизонтального масштабирования.
Виртуальный или физический экземпляр сервера.
Набор узлов, которые составляют отдельную группу.
Узел, который хранит подмножество данных из набора.
Набор реплик
Ряд узлов, на которых хранятся копии набора данных. У каждого хранилища в наборе реплик есть роль: мастер или реплика.
Хранилище в наборе реплик, которое обрабатывает запросы на чтение и запись.
Хранилище в наборе реплик, которое обрабатывает только запросы на чтение.
Запросы на чтение
Запросы только на чтение, то есть выборка.
Запросы на запись
Операции по изменению данных, то есть запросы на создание, замену, обновление и удаление данных.
Сегменты (виртуальные сегменты)
Абстрактные виртуальные узлы, на которые производится секционирование набора данных по сегментному ключу (идентификатору сегмента).
Идентификатор сегмента
A sharding key defining which bucket belongs to which replica set. A bucket id may be calculated from a hash key.
Прокси-сервер, который отвечает за запросы маршрутизации от приложения к узлам в кластере.