База по шардированию базы

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

Мы на пальцах рассмотрим что такое шардирование, как оно помогает в масштабировании и даже рассмотрим тот самый этап «роста».

О чём речь?

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

Шардирование помогает оптимизировать хранение данных приложения за счёт их распределения между инсталляциями БД (которые находятся на разных железках), что улучшает отзывчивость сервиса, так как размер данных в целом на каждом инстансе станет меньше.

Шардирование — это разновидность партиционирования (от англ. partition — деление, раздел). Отличие в том, что партиционирование подразумевает разделение данных внутри одной БД, а шардирование распределяет их по разным экземплярам БД.

Способы шардирования

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

Средствами БД. Некоторые базы — MongoDB, Elasticsearch, ClickHouse и другие — умеют самостоятельно распределять данные между своими экземплярами, для этого достаточно настроить конфигурацию. На мой взгляд, это лучший вариант.
Надстройками к БД. Самый спорный способ — применение надстроек, которые выполняют шардирование, например Vitess или Citus, поскольку при этом есть риск потери данных и производительности.
Клиентскими средствами. В этом случае экземпляры БД даже не подозревают о существовании друг друга, шардированием управляет стороннее приложение — со всеми вытекающими рисками.

Методы работы в этих способах схожи: мы выбираем ключ для распределения данных (это может быть идентификатор, временная метка или хеш записи) и в соответствии с ним записываем информацию в нужный шард. Как правило, ключи стараются выбирать так, чтобы данные были равномерно распределены по шардам. Сделать это не сложно — достаточно ориентироваться на текущее содержимое БД.

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

Пример шардирования

Давайте в качестве примера сделаем клиентское шардирование горячо любимой в Ozon PostgreSQL. Приложение будет на Go, а мигрировать будем с помощью Goose. Для начала нам надо добавить сами шарды, то есть развернуть еще одну инсталляцию БД. Отвлекаться на детальный разбор того, как правильно раскатывать PostgreSQL, мы не будем.

Добавим в наш Storage маппинг шардов:

// Обозначим количество шардов.
const bucketQuantity = 2

const (
    Shard1 ShardNum = iota
    Shard2
)

// Для лучшей семантики.
type ShardNum int
type shardMap map[ShardNum]*sqlx.DB

type Storage struct {
    shardMap shardMap
}

Напишем конструктор для Storage который возьмёт на себя все задачи по инициализации соединений с БД.

func initShardMap(ctx context.Context, dsns map[ShardNum]string) shardMap {
    m := make(shardMap, len(dsns))
    for sh, dsn := range dsns {
               m[sh] = discoveryShard(ctx, dsn)
    }

    return m
}

func discoveryShard(ctx context.Context, dsn string) *sqlx.DB {
    db, err := sqlx.ConnectContext(ctx, "postgres", dsn)
    if err != nil {
               panic(err)
    }

    return db
}

func NewStorage(ctx context.Context, dsns map[ShardNum]string) *Storage {
    return &Storage{
               shardMap: initShardMap(ctx, dsns),
    }
}

Переходим к работе с данными. Реализуем методы для их записи в шарды и чтения оттуда. Начинается всё с определения того, в какой шард идти.

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

func (s *Storage) shardByItemID(itemID int64) ShardNum {
   return ShardNum(itemID % bucketQuantity)
}

У нас есть вот такой незаурядный метод чтения из БД. Тут стоит обратить внимание на то, что мы выполняем запрос на инстансе БД из нашего маппинга, а получаем инстанс (*sqlx.DB) по идентификатору шарда из сигнатуры.

func (s *Storage) getItemsByID(ctx context.Context, shard *sqlx.DB, itemsIDs []int64) ([]models.Item, error) {
   items := make([]models.Item, 0)

   query, args, err := sq.
       Select(itemsTableFields...).
       From(itemsTable).
       Where(sq.Eq{itemIDField: itemsIDs}).
       PlaceholderFormat(sq.Dollar).
       ToSql()
   if err != nil {
       err = errors.Wrap(err, "[create query]")
       return items, err
   }

   err = shard.SelectContext(ctx, &items, query, args...)
   return items, err
}

Сам идентификатор шарда мы получаем чуть выше, когда распределяем наши ItemIDs по кубышкам. Само распределение выглядит вот так:

func (s *Storage) sortItemsIDsByShard(itemIDs ...int64) map[ShardNum][]int64 {
   shardToItems := make(map[ShardNum][]int64)

   for _, id := range itemIDs {
       shardID := s.shardByItemID(id)
       if _, ok := shardToItems[shardID]; !ok {
           shardToItems[shardID] = make([]int64, 0)
       }

       shardToItems[shardID] = append(shardToItems[shardID], id)
   }

   return shardToItems
}

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

func (s *Storage) GetItems(ctx context.Context, itemIDs ...int64) ([]models.Item, error) {
   shardToItems := s.sortItemsIDsByShard(itemIDs...)

   respChan := make(chan []models.Item, len(shardToItems))
   errChan := make(chan error, len(shardToItems))
   wg := &sync.WaitGroup{}

   for shardID, ids := range shardToItems {
       wg.Add(1)
       shard := s.shardMap[shardID]
       go s.asyncGetItemsByID(ctx, shard, ids, wg, respChan, errChan)
   }

   wg.Wait()
   close(respChan)
   close(errChan)

   result := make([]models.Item, 0)
   for items := range respChan {
       result = append(result, items...)
   }

   errs := make([]error, 0, len(errChan))
   for e := range errChan {
       errs = append(errs, e)
   }

   err := multierr.Combine(errs...)

   return result, err
}

Для того чтобы не терять смысл getItemsByID за нагромождением каналов и Wait-групп, мы просто обернем всё это в asyncGetItemsByID:

unc (s *Storage) asyncGetItemsByID(
   ctx context.Context,
   shard *sqlx.DB,
   itemsIDs []int64,
   wg *sync.WaitGroup,
   resp chan<- []models.Item,
   errs chan<- error,
) {
   defer wg.Done()
   items, err := s.getItemsByID(ctx, shard, itemsIDs)
   if err != nil {
       errs <- errors.Wrapf(err, "[getItemsByID] can't select from shard %d", shard)
   }

   resp <- items
}

Всё то же самое мы проделываем для записи данных в шарды:

Ну и скриптик для миграции всего этого дела:

#!/usr/bin/env bash
export MIGRATION_DIR=./migrations/

if [ "${STAGE}" = "production" ]; then
   if [ "$1" = "--dryrun" ]; then
       goose -dir ${MIGRATION_DIR} postgres "user=${USER1} password=${PASSWORD1} dbname=${DBNAME1} host=${HOST1} port=${PORT1} sslmode=disable" status
       goose -dir ${MIGRATION_DIR} postgres "user=${USER2} password=${PASSWORD2} dbname=${DBNAME2} host=${HOST2} port=${PORT2} sslmode=disable" status
   else
       goose -dir ${MIGRATION_DIR} postgres "user=${USER1} password=${PASSWORD1} dbname=${DBNAME1} host=${HOST1} port=${PORT1} sslmode=disable" up
       goose -dir ${MIGRATION_DIR} postgres "user=${USER2} password=${PASSWORD2} dbname=${DBNAME2} host=${HOST2} port=${PORT2} sslmode=disable" up
   fi

elif [ "${STAGE}" = "staging" ]; then
   if [ "$1" = "--dryrun" ]; then
       goose -dir ${MIGRATION_DIR} postgres "user=${USER1} password=${PASSWORD1} dbname=${DBNAME1} host=${HOST1} port=${PORT1} sslmode=disable" status
   else
       goose -dir ${MIGRATION_DIR} postgres "user=${USER1} password=${PASSWORD1} dbname=${DBNAME1} host=${HOST1} port=${PORT1} sslmode=disable" up
   fi

elif [ "${STAGE}" = "development" ]; then
   exit 0
fi

Очень удобно шардировать в приложении, где еще нет данных, а следовательно нет необходимости их перетаскивать. Но что делать, если мы шардим рабочее приложение? Тут, как говорится у нас на Руси, case-by-case.

Приложение можно остановить? Прекрасно! Останавливаем разбор очередей, отключаем обработку запросов или вовсе останавливаем приложение, делаем резервную копию базы, если ещё не сделали, перетаскиваем данные в соответствии с выбранным ключом и снова вводим приложение в эксплуатацию, предварительно проведя регрессионное тестирование.

Приложение должно отвечать на запросы? Тогда делаем временную надстройку внутри репозитория. Пишем данные всегда в новые шарды. Читаем сначала из нового шарда; если там нет нужных данных, то обращаемся к старым шардам. Если данные оказываются в старом шарде, то при желании можно их переносить в новый — и в конце концов данные перераспределятся между шардами. Только не забудьте добавить метрики, чтобы не пропустить этот знаменательный момент. И обязательно проверьте, все ли данные переехали или только те, к которым идут запросы. Да, и на первых порах это не то чтобы положительно скажется на отзывчивости приложения, будьте к этому готовы.
Если приложение пишет данные в БД только по событиям из условной kafka, а синхронные запросы (REST/GRPC) только читающие (классическая ситуация для event sourcing), то мы отключаем чтение из kafka, выкатываем в prod инстанс версии приложения, которое уже живет с новой схемой шардов, но синхронные запросы шлем только на инстанс приложения предыдущей версии (оно же canary-deploy). Далее джоба внутри приложения последовательно читает данные по старому маппингу, и пишет в новый, после переноса можно сразу же и удалять данные в старой схеме.

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

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

Начинается всё с создания дублирующей схемы шардов внутри Storage. Внесём изменения в константы:

const legacyBucketQuantity = 2
const bucketQuantity = 3

const (
   Shard1 ShardNum = iota
   Shard2
   Shard3
)

Заведём внутри Storage shardMapLegacy, который содержит дорешардинговый маппинг:

type Storage struct {
    shardMapLegacy shardMap
    shardMap      shardMap
}

Ну и инициализация. В конструктор Storage теперь будем передавать также две схемы:

func NewStorage(ctx context.Context, dsns map[ShardNum]string, dsnsLegacy map[ShardNum]string) *Storage {
   return &Storage{
       shardMap:       initShardMap(ctx, dsns),
       shardMapLegacy: initShardMap(ctx, dsnsLegacy),
   }
}

Заводим метод для получения shardID, чтобы после переноса данных его удалить:

func (s *Storage) legacyShardByItemID(itemID int64) ShardNum {
    return ShardNum(itemID % legacyBucketQuantity)
}

Ну и ещё чуть-чуть дублирования кода. Речь о практически полной копии sortItemsIDsByShard; разница лишь в том, что для получения идентификатора шарда мы используем ранее модифицированную функцию.

func (s *Storage) sortItemsIDsByLegacyShard(itemIDs ...int64) map[ShardNum][]int64 {
    shardToItems := make(map[ShardNum][]int64)

    for _, id := range itemIDs {
               shardID := s.legacyShardByItemID(id)
               if _, ok := shardToItems[shardID]; !ok {
                             shardToItems[shardID] = make([]int64, 0)
               }

             shardToItems[shardID] = append(shardToItems[shardID], id)
    }

    return shardToItems
}

Метод добавления Items в изменении не нуждается, так как мы условились, что данные пишем всегда в свежие шарды, а вот GetItems надо подправить. Теперь он будет конкурентно выполнять запрос сразу в две схемы, а полученные данных мы будем склеивать, отдавая предпочтение данным с актуального маппинга шардов.

func (s *Storage) GetItems(ctx context.Context, itemIDs ...int64) ([]models.Item, error) {
   wg := &sync.WaitGroup{}

   resultLegacy := make([]models.Item, 0)
   resultActual := make([]models.Item, 0)

   var err error

   wg.Add(1)
   go func() {
       defer wg.Done()

       res, e := s.getItems(ctx, itemIDs...)
       err = multierr.Append(err, e)
       resultActual = res
   }()

   wg.Add(1)
   go func() {
       defer wg.Done()

       res, e := s.getItemsFromLegacyShardMap(ctx, itemIDs...)
       err = multierr.Append(err, e)
       resultLegacy = res
   }()

   wg.Wait()
   result := mergeItems(resultActual, resultLegacy)

   return result, err
}

Склейка результата выглядит так:

func mergeItems(items, legacyItems []models.Item) []models.Item {
   itemsMap := make(map[models.Item]struct{})

   for _, item := range legacyItems {
       itemsMap[item] = struct{}{}
   }

   for _, item := range items {
       itemsMap[item] = struct{}{}
   }

   mergedItems := make([]models.Item, 0, len(itemsMap))
   for item, _ := range itemsMap {
       mergedItems = append(mergedItems, item)
   }

   return mergedItems
}

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

При таком подходе остаётся только один вопрос: что делать с «мёртвыми» данными, которые лежат не в своих шардах после решардинга?

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

DELETE FROM items
WHERE ctid IN (
   SELECT ctid
   FROM items
   WHERE id % 4 NOT IN (2, 2-4)
   );

З.Ы. Вариант SQL-запроса предполагает, что мы храним гошный UInt64 в постгревом BigInt. В этом случае положительные гошные числа могут превратиться в отрицательные постгревые, поэтому делаем NOT IN для ренджа.

Иногда встречаются системы, где данные имеют свойства ‎‎‎‎«протухать». В таких системах самое логичное оставить данные после решардинга, и дождаться пока они «протухнут».

И пара слов об упячках, с которыми я сталкивался, — о партиционировании внутри одной БД и шардинге целыми партициями. Поначалу кажется, что это логично и даже элегантно. Ведь для решардинга достаточно просто перетащить целую партицию с одного шарда в другой. И это ФАТАЛЬНАЯ ОШИБКА. Со временем вы устанете от трёхэтажного мата негодования, вызванного пятиэтажными пакетными запросами, из-за которых горячие данные не будут нормально попадать в кеш. Такой способ работает лишь в том случае, если партиционирование выполняется по дате, но запросы, как правило, обращаются к свежим или старым данным, как, например, во многих OLAP-системах. В остальных случаях перспективнее держать данные в рамках одной партиции, а решардить их путём постепенного переноса, если, конечно, БД не предусматривает своих вариантов решения проблемы решардинга.

Вместо вывода

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

Клиентское шардирование — надо.

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

Не так страшно шардирование, как решардинг. Важно заранее подумать о том, как вы будете решать вопросы консистентности при решардинге.

← Предыдущая статья

Что такое облачная база данных?