Шардинг

Шардинг позволяет реализовать горизонтальное масштабирование баз данных — распределение набора данных по нескольким узлам (шардам). В Reindexer возможно разделение неймспейса по нескольким шардам или хранение неймспейсов БД целиком на разных серверах.

Общее описание шардинга в Reindexer и особенности реализации

В настоящее время Reindexer поддерживает единственный режим — шардинг по значениям индекса (ключа шардирования). При этом:

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

  • Неймспейс может иметь только один ключ шардирования.

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

  • Максимальное количество шард, поддерживаемое Reindexer, — 999.

В качестве ключа шардирования рекомендуется использовать индекс, являющийся частью композитного PK-индекса. Это позволяет избежать конфликтов PK-индексов в ходе «решардинга» (изменения механизма распределения данных между шардами).

Пример для демонстрации работы шардинга

Пример скрипта для демонстрации работы шардинга в Reindexer находится здесь. Этот скрипт устанавливает и запускает три экземпляра сервера Reindexer, которые сконфигурированы как кластер из трёх шардов. В конфигурации настроено шардирование записей неймспейса ns по ключу location (сам неймспейс и индексы нужно создать вручную). Shard0 с адресом cproto://127.0.0.1:6110 установлен как узел по умолчанию.

Проксирование запросов

Каждый запрос выполняется, либо на одном шарде, либо на всех шардах. Выполнение запросов на подмножествах шардов на данный момент не поддерживается.

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

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

SELECT * FROM ns WHERE location='name1' and id > 100

В этом случае запрос будет выполнен на шарде, где хранятся документы со значением ключа шардирования name1.

Распределенные запросы также поддерживаются. При этом есть некоторые ограничения для агрегирования. Например, такой запрос вернёт записи с id > 100 со всех шард:

SELECT * FROM ns WHERE and id > 100

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

SELECT * FROM sharded_ns1 JOIN (SELECT * FROM sharded_ns2 WHERE location='name1') ON sharded_ns1.id=sharded_ns2.id WHERE location='name1'

Также возможно выполнения JOINED-запросов с присоединением локальных неймспейсов шарда, которые не указаны в файле конфигурации шардинга:

SELECT * FROM sharded_ns JOIN (SELECT * FROM not_sharded_ns) ON sharded_ns.id=not_sharded_ns.id WHERE location='name1'

Локальный неймспейс шарда может использоваться в JOINED-запросе в качестве левого неймспейса. Такой запрос будет проксироваться на указанный в joined-подзапросе шард:

SELECT * FROM not_sharded_ns JOIN (SELECT * FROM sharded WHERE location='name1') ON sharded_ns.id=not_sharded_ns.id

Выполнение локальных запросов на шардах

Для локального выполнения запроса на шарде, без проксирования на другие, используйте оператор LOCAL. Примеры:

LOCAL SELECT * FROM sharded_ns1 JOIN (SELECT * FROM sharded_ns2) ON sharded_ns1.id=sharded_ns2.id;
LOCAL SELECT * FROM sharded_ns JOIN (SELECT * FROM not_sharded_ns) ON sharded_ns.id=not_sharded_ns.id;
LOCAL SELECT * FROM not_sharded_ns JOIN (SELECT * FROM sharded_ns) ON sharded_ns.id=not_sharded_ns.id;

Оператор LOCAL может использоваться только с SELECT-запросами.

Настройка и использование шардинга

Reindexer поддерживает 2 способа конфигурирования шардинга: с помощью файлов конфигурации и в рантайме через служебный неймспейс #config с использованием механизма action (при помощи reindexer_tool, RPC, gRPC или REST API).

Настройка шардинга через файлы конфигурации

Шардинг конфигурируется с помощью файлов replication.conf и sharding.conf. Reindexer считывает их содержимое при запуске.

В файле replication.conf указываются общие параметры репликации. Он синхронизируется со служебным неймспейсом #config: то, что записано в файл, будет перенесено в неймспейс, и наоборот. При этом, если файла нет, то создаваться автоматически он не будет. В replication.conf параметр server_id для каждого шарда должен быть уникальным. Параметры из файла replication.conf описаны в разделе «Настройка общих параметров репликации».

В файле sharding.conf содержатся специфические параметры шардинга. Он также синхронизируется со служебным неймспейсом #config.

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

Здесь указываются параметры:

  • namespaces - список неймспейсов для шардинга,

    • namespace — имя неймспейса,
    • default_shard — id шарда по умолчанию. На нем будут храниться все записи со значениями ключа шардирования, которые не заданы в условиях для шардинга на другие шарды.
    • index — индекс, по которому будет производиться распределение записей по шардам,
      • keys — ключи шардирования,

        • shard_id — id шарда, на котором будут храниться записи с заданным значением ключа,

        • values — значения ключа шардирования для распределения записей на шард с указанным выше id, могут быть представлены:

          • в виде отдельных значений,
          • в виде диапазонов: [100, 400]. Границы диапазонов включены.
          • в виде комбинации отдельных значений и диапазонов (пересечения в рамках одного шарда допустимы):
          keys:
   - shard_id: 1
      values:
         - 0
         - 1
         - 2
         - 15
         - [10, 100]
         - [300, 10000]
   - shard_id: 2
      values:
         - [150, 200]
   - shard_id: 3
      values:
         - 11

  • shards — сведения об узлах, участвующих в шардинге,

    • shard_id — id узла,
      • dsns — адрес узла (или несколько адресов, если шардирование используется совместно с кластеризацией),

  • this_shard_id — id данного шарда. Должен быть уникальным для каждого шарда,

  • reconnect_timeout_msec — таймаут для переподключения для одного узла (лимит времени для запроса статуса),

  • shards_awaiting_timeout_sec — время ожидания подключения шардов при выполнении распределённых операций (например, OpenNamespace или AddIndex). Параметр необходим, т.к. некоторые операции требуют, чтобы все шарды были онлайн и при этом они могут запускаться не одновременно. Значение 0 для этого параметра отключает механизм ожидания, -1 — устанавливает неограниченное время ожидания.

  • proxy_conn_count — количество прокси-соединений между шардами. Диапазон возможных значений — от 1 до 64.

  • proxy_conn_concurrency — количество одновременных запросов для каждого прокси-соединения. Диапазон возможных значений — от 1 до 1024.

  • proxy_conn_threads — количество потоков для прокси-соединений.

  • source_id — параметр, который используется для определения источника конкретной конфигурации, а также для нахождения наиболее свежих конфигураций шардинга при синхронизации raft-кластера. Значение поля генерируется в момент задания конфигурации на основе текущего времени и server ID. Это поле не стоит добавлять или редактировать при изменении конфигурации шардинга вручную. Обычно оно добавляется в результате выполнения команды apply_sharding_config.

Настройка шардинга в рантайме

Настройка шардинга в рантайме возможна в случае его самостоятельного использования, а также при совместном применении шардинга и синхронной репликации. В обоих случаях случае используется action apply_sharding_config для внесения изменений в служебный неймспейс #config.

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

Конфигурация шардинга применяется синхронно: если запрос с командой apply_sharding_config вернул ОК, значит на все ноды новая конфигурация была установлена успешно.

При задании конфигурации через apply_sharding_config на всех узлах, участвующих в шардинге, source_id будет совпадать. В случае неконсистентных source_id распределённые запросы будут возвращать ошибку.

ВАЖНО При настройке шардинга в рантайме используйте одну из последовательностей действий:

  1. Сначала примените новую конфигурацию, а после этого в соответствии с ней создайте неймспейсы и индексы на нодах, участвующих в шардинге.
  2. Сначала создайте на всех нодах неймспейсы и индексы для шардинга, а затем примените новую конфигурацию, в которой будут указаны эти неймспейсы и индексы.

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

Примеры команд для изменения настроек шардинга в рантайме через reindexer_tool при использовании утилиты в интерактивном режиме:


 \upsert #config {"type":"action","action":{"command":"apply_sharding_config","config": {"version":1,"namespaces":[{"namespace":"ns","default_shard":0,"index":"location","keys":[{"shard_id":0,"values":["Africa"]},{"shard_id":1,"values":["Europe","Asia"]},{"shard_id":2,"values":["North America","Australia"]}]}],"shards":[{"shard_id":0,"dsns":["cproto://127.0.0.1:6110/sharding_db"]},{"shard_id":1,"dsns":["cproto://127.0.0.1:6210/sharding_db"]},{"shard_id":2,"dsns":["cproto://127.0.0.1:6310/sharding_db"]}],"this_shard_id":-1} }}

json-представление передаваемой в команде конфигурации:

{
  "version": 1,
  "namespaces": [
    {
      "namespace": "ns",
      "default_shard": 0,
      "index": "location",
      "keys": [
        {
          "shard_id": 0,
          "values": [
            "Africa"
          ]
        },
        {
          "shard_id": 1,
          "values": [
            "Europe",
            "Asia"
          ]
        },
        {
          "shard_id": 2,
          "values": [
            "North America",
            "Australia"
          ]
        }
      ]
    }
  ],
  "shards": [
    {
      "shard_id": 0,
      "dsns": [
        "cproto://127.0.0.1:6110/sharding_db"
      ]
    },
    {
      "shard_id": 1,
      "dsns": [
        "cproto://127.0.0.1:6210/sharding_db"
      ]
    },
    {
      "shard_id": 2,
      "dsns": [
        "cproto://127.0.0.1:6310/sharding_db"
      ]
    }
  ],
  "this_shard_id": -1
}


\upsert #config {"type":"action","action":{"command":"apply_sharding_config","config": {"version":1,"namespaces":[{"namespace":"ns","default_shard":0,"index":"location","keys":[{"shard_id":0,"values":["Africa"]},{"shard_id":1,"values":["Europe","Asia"]},{"shard_id":2,"values":["North America","Australia"]}]}],"shards":[{"shard_id":0,"dsns":["cproto://127.0.0.1:7000/base_test","cproto://127.0.0.1:7001/base_test","cproto://127.0.0.1:7002/base_test"]},{"shard_id":1,"dsns":["cproto://127.0.0.1:7010/base_test","cproto://127.0.0.1:7011/base_test","cproto://127.0.0.1:7012/base_test"]},{"shard_id":2,"dsns":["cproto://127.0.0.1:7020/base_test","cproto://127.0.0.1:7021/base_test","cproto://127.0.0.1:7022/base_test"]}],"this_shard_id":-1} }}

json-представление передаваемой в команде конфигурации:

{
  "version": 1,
  "namespaces": [
    {
      "namespace": "ns",
      "default_shard": 0,
      "index": "location",
      "keys": [
        {
          "shard_id": 0,
          "values": [
            "Africa"
          ]
        },
        {
          "shard_id": 1,
          "values": [
            "Europe",
            "Asia"
          ]
        },
        {
          "shard_id": 2,
          "values": [
            "North America",
            "Australia"
          ]
        }
      ]
    }
  ],
  "shards": [
    {
      "shard_id": 0,
      "dsns": [
        "cproto://127.0.0.1:7000/base_test",
        "cproto://127.0.0.1:7001/base_test",
        "cproto://127.0.0.1:7002/base_test"
      ]
    },
    {
      "shard_id": 1,
      "dsns": [
        "cproto://127.0.0.1:7010/base_test",
        "cproto://127.0.0.1:7011/base_test",
        "cproto://127.0.0.1:7012/base_test"
      ]
    },
    {
      "shard_id": 2,
      "dsns": [
        "cproto://127.0.0.1:7020/base_test",
        "cproto://127.0.0.1:7021/base_test",
        "cproto://127.0.0.1:7022/base_test"
      ]
    }
  ],
  "this_shard_id": -1
}

Значение параметра this_shard_id в представленных выше примерах конфигураций не важно. Оно будет рассчитано автоматически в соответствии со значениями в shards для каждой ноды и добавлено в ее конфигурацию.

Настройка шардинга в рантайме для одной конкретной ноды

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

Чтобы применить локальные настройки шардинга для одной конкретной ноды, при их передаче используйте флаг "locally": true в команде apply_sharding_config.

При применении локальных настроек в команде apply_sharding_config также необходимо передавать актуальный source_id, заданный вручную. Иначе он будет сгенерирован автоматически для данной ноды и не совпадёт с source_id других шард, из-за чего возникнет неконсистентность source_id и распределённые запросы будут возвращать ошибку.

Пример команды для применения локальных настроек шардинга в рантайме через reindexer_tool для одной конкретной ноды:


 \upsert #config {"type":"action","action":{"command":"apply_sharding_config","locally":true,"source_id":19719893309960188,"config":{"version":1,"namespaces":[{"namespace":"ns","default_shard":0,"index":"location","keys":[{"shard_id":0,"values":["Africa"]},{"shard_id":1,"values":["Europe","Asia"]},{"shard_id":2,"values":["North America","Australia"]}]}],"shards":[{"shard_id":0,"dsns":["cproto://127.0.0.1:6110/sharding_db"]},{"shard_id":1,"dsns":["cproto://127.0.0.1:6210/sharding_db"]},{"shard_id":2,"dsns":["cproto://127.0.0.1:6310/sharding_db"]}],"this_shard_id":-1}}}

json-представление передаваемой в команде конфигурации:

{
  "type": "action",
  "action": {
    "command": "apply_sharding_config",
    "locally": true,
    "source_id": 19719893309960188,
    "config": {
      "version": 1,
      "namespaces": [
        {
          "namespace": "ns",
          "default_shard": 0,
          "index": "location",
          "keys": [
            {
              "shard_id": 0,
              "values": [
                "Africa"
              ]
            },
            {
              "shard_id": 1,
              "values": [
                "Europe",
                "Asia"
              ]
            },
            {
              "shard_id": 2,
              "values": [
                "North America",
                "Australia"
              ]
            }
          ]
        }
      ],
      "shards": [
        {
          "shard_id": 0,
          "dsns": [
            "cproto://127.0.0.1:6110/sharding_db"
          ]
        },
        {
          "shard_id": 1,
          "dsns": [
            "cproto://127.0.0.1:6210/sharding_db"
          ]
        },
        {
          "shard_id": 2,
          "dsns": [
            "cproto://127.0.0.1:6310/sharding_db"
          ]
        }
      ],
      "this_shard_id": -1
    }
  }
}

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

Условия успешного применения настроек шардинга при конфигурировании в рантайме

Конфигурация шардинга в рантайме будет успешно применена при выполнении следующих условий:

  1. Если RAFT-кластер не используется, то все ноды из текущего и нового конфига шардинга должны быть доступны.

  2. Шардируемый неймспейс из новой конфигурации должен соответствовать одному из условий:

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

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

    • Для RAFT-кластера необходим консенсус.
    • Все ноды, перечисленные в разделе shards для конкретного shard_id, должны быть нодами одного синхронного кластера и в нём не должно быть узлов, не являющихся частью шарда. Либо шард должен являться отдельным узлом, не входящим в синхронный кластер.

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

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

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

   shard0           shard1
   
   node0------------node3
   /    \      ___/ /   \
node1--node2__/  node4--node5

Для реализации подобной схемы каждый узел должен иметь уникальный server_id, а каждый RAFT-кластер может (но не должен) иметь уникальный cluster_id. Узлы из различных шард автоматически выбирают DSN из списка, указанного в файле конфигурации, пытаясь подключиться к лидеру RAFT-кластера.

Изменение способа распределения данных между шардами

Единственный способ изменить распределение данных между шардами — использование механизма дампа/восстановления в reindexer_tool. Ниже приведены примеры создания дампа из одного шардингового кластера (cproto://127.0.1:6534/mydb) и его восстановления в другом кластере (cproto:/127.0.0.1:7231/mydb) в разных режимах:

При использовании режима sharded_only создается дамп, содержащий только шардируемые неймспейсы с текущего шарда и других узлов кластера. Пример:

reindexer_tool --dsn cproto://127.0.0.1:6534/mydb --command '\dump' --dump-mode=sharded_only --output mydb.rxdump
reindexer_tool --dsn cproto://127.0.0.1:7231/mydb -f mydb.rxdump

Если второй кластер из приведенного примера будет иметь другую конфигурацию шардинга в файле sharding.conf, все шардируемые данные будут по-новому распределены между шардами. Режим будет полезен, если нужно изменить схему распределения данных в шардируемых неймспейсах.

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

reindexer_tool --dsn cproto://127.0.0.1:6534/mydb --command '\dump' --dump-mode=full_node --output mydb.rxdump
reindexer_tool --dsn cproto://127.0.0.1:7231/mydb -f mydb.rxdump

При использовании режима local_only создается дамп, содержащий только данные из локальных неймспейсов текущего узла. Данные из шардируемых неймспейсов в дамп не попадают.

reindexer_tool --dsn cproto://127.0.0.1:6534/mydb --command '\dump' --dump-mode=local_only --output mydb.rxdump
reindexer_tool --dsn cproto://127.0.0.1:7231/mydb -f mydb.rxdump

Особенности и ограничения шардинга в Reindexer

  1. Reindexer не поддерживает автоматический перенос документа с одного шарда на другой. Это значит, что значение поля ключа шардинга не должно изменяться с помощью UPDATE или UPSERT запросов — их использование для этого может привести к неожиданному поведению. Если же вам необходимо изменить значение поля ключа шардирования и перенести запись на другой шард, удалите ее и добавьте в неймспейс заново с требуемым новым значением.

  2. Метаданные являются общими для всех шард. Исключение — внутренние системные метазаписи.

  3. Запросы на обновление и удаление данных (UPDATE/DELETE) могут выполняться только на одном шарде с обязательным указанием ключа шардирования. Например, для удаления записи с id = 1, нужно выполнить запрос вида DELETE FROM ns where location = 'Europe' and id = 1 (где location — ключ шардирования).

  4. Распределенные запросы по полнотекстовому индексу не поддерживаются.

  5. Для ключей шардирования в запросах допускается использование только условия =. Таким образом, каждый запрос выполняется на одном узле, либо на каждом узле (если ключ шардирования в запросе не указан).

  6. Распределенные объединения и агрегации AVG, Facet и Distinct не поддерживаются.

  7. Полнотекстовые, композитные и Array-индексы не могут использоваться в качестве ключа шардирования.

  8. EXPLAIN не поддерживается для распределенных запросов.

  9. EXPLAIN на данный момент не показывает сведения о времени проксирования запросов на шард.

  10. Protobuf/MsgPack через GRPC пока не поддерживается в настройках Reindexer для кластера/шардинга.