Курс Модуль 4. Работа с БД - Лекция: Шардирование: обратная сторона

2.1 Как зашардить и втормозить в N раз?

Можно так зашардить и втормозить ровно в N раз:

Послать запросы docs00...docs15 последовательно, а не параллельно.
В простых запросах сделать выборку не по ключу, WHERE something=234.

В этом случае,сериализованная часть (serial) занимает не 1% и не 5%, а примерно 20% в современных базах данных. Можно получить и 50% сериализованной части, если обращаться к базе данных по дико эффективному бинарному протоколу или линковать ее как динамическую библиотеку в скрипт на Python.

Все остальное время обработки простого запроса будут занимать не распараллеливаемые операции разбора запроса, подготовки плана и т.д. То есть тормозит не чтение записи.

Если мы разобьем данные на 16 таблиц и будем запускать последовательно, как принято в языке программирования PHP, например, (он не очень хорошо умеет запускать асинхронные процессы), то как раз и получим замедление в 16 раз. А, возможно, даже больше, потому что добавятся еще и network round-trips.

Внезапно, при шардировании важен выбор языка программирования.

Помним про выбор языка программирования, потому что если посылать запросы к базе данных (или поисковому серверу) последовательно, то откуда взяться ускорению? Скорее, появится замедление.

2.2 Про полуавтомат

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

Страдающее человечество в лице отдельно взятых DBA мучается годами и пишет несколько плохих решений для шардинга, построенных непонятно на чем. После этого пишется одно более-менее приличное решение для шардинга под названием ProxySQL (MariaDB/Spider, PG/pg_shard/Citus, ...). Это хорошо известный пример этой самой нашлёпки.

ProxySQL в целом, конечно же, полноценное решение enterprise-класса для open source, для роутинга и прочего. Но одна из решаемых задач — это шардинг для базы данных, которая сама по себе шардить по-человечески не умеет. Понимаете, нет рубильника «shards=16», приходится либо в приложении переписывать каждый запрос, а их местами много, либо между приложением и базой данных ставить некий промежуточный слой, который смотрит: «Хм… SELECT*FROM documents? Да его надо разорвать на 16 маленьких SELECT*FROM server1.document1, SELECT*FROM server2.document2 — к этому серверу с таким логином/паролем, к этому с другим. Если один не ответил, то...» и т.д. Ровно этим могут заниматься промежуточные нашлёпки. Они есть чуть менее, чем для всех баз данных. Для PostgreSQL, насколько я понимаю, одновременно есть и какие-то встроенные решения (PostgresForeign Data Wrappers, по-моему, встроен в сам PostgreSQL), есть внешние нашлёпки.

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

2.3 Абсолютная идеальная автоматика?

Вся теория кайфа в случае шардирования в этой букве F(), базовый принцип всегда один и тот же грубо: shard_id = F(object).

Шардирование — это вообще про что? У нас есть 2 миллиарда записей (или 64). Мы их хотим раздробить на несколько кусков. Возникает неожиданный вопрос — как? По какому принципу я свои 2 миллиарда записей (или 64) должен разбросать на доступные мне 16 серверов?

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

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

shard_func = F1(object); 
shard_id = F2(shard_func, ...); 
shard_id = F2(F1(object), current_num_shards, ...).

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

2.4 Какие бывают F()?

Их можно придумать много разных и много разных механизмов реализации. Примерная краткая сводка:

F = rand() % nums_shards
F = somehash(object.id) % num_shards
F = object.date % num_shards
F = object.user_id % num_shards
...
F = shard_table [ somehash() |… object.date |… ]

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

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

F = rand()

Раскидывать радомом — не очень правильный метод. Одна проблема: раскидали мы наши 2 млрд записей на тысячу серверов случайным образом, и не знаем, где запись лежит. Нам надо вытащить user_1, а где он, не знаем. Идем на тысячу серверов и перебираем все — как-то это неэффективно.

F = somehash()

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

А зачем мы это делаем? А затем, что у нас highload и в один сервер у нас больше ничего не влазит. Если бы влазило, жизнь была бы такая простая.

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

Естественное шардирование (F = object.date % num_shards)

Иногда, то есть часто, 95% трафика и 95% нагрузки — это запросы, у которых есть какое-то естественное шардирование. Например, 95% условно социально-аналитических запросов затрагивает данные только за последние 1 день, 3 дня, 7 дней, а оставшиеся 5% обращаются к нескольким последним годам. Но 95% запросов, таким образом, естественно шардированы по дате, интерес пользователей системы сфокусирован на последних нескольких днях.

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

Жизнь улучшается — мы теперь не только знаем расположение конкретного объекта, но и про диапазон тоже знаем. Если у нас спрашивают не диапазон дат, а диапазон других колонок, то, конечно, придется перебирать все шарды. Но по условиям игры у нас всего 5 % таких запросов.

Вроде бы мы придумали идеальное решение всего, но есть две проблемы:

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

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

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

2.5 Цена, которую нужно заплатить

Формально мы знаем теперь знаем «всё». Правда, мы не знаем одну гигантскую головную боль и две головные боли поменьше.

1. Простая боль: плохо размазало

Это пример из учебника, который в бою почти не встречается, но вдруг.

Как пример с датой, только без даты!
Ненамеренное неравномерное (ощутимо) распределение.

Выбрали механизм шардирования, и/или данные поменялись, и, конечно же, PM не донес требования (у нас же не бывает ошибок в коде, всегда PM требования не доносит), и распределение стало чудовищно неравномерным. То есть промазали с критерием.

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

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

2. «Непобедимая» боль: агрегация, join

Как делать выборки, которые джойнят миллиард записей из одной таблицы на миллиард записей из другой таблицы?

Как «быстро» посчитать… WHERE randcol BETWEEN aaa AND bbb?
Как «ловко» сделать… users_32shards JOIN posts_1024 shards?

Короткий ответ: никак, страдать!

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

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

Это отдельный курс лекций на три дня, поэтому переходим к последней адской боли и к разным алгоритмам борьбы с ней.

2.6. Сложная/длинная боль: решардинг

Готовьтесь: если вы зашардили ваши данные первый раз в жизни, то в среднем еще раз пять вы их порешардите обязательно.

Сколько кластер не конфигурируй, всё равно решардить.

Если вы очень умны и удачливы, то перешардите, минимум, один раз. Но один раз вы обязательно, потому что в тот момент, когда вы думаете, что пользователю достаточно 10 единиц, кто-то прямо в этот момент пишет запрос на 30, а в планах имеет запрос на 100 единиц неизвестных ресурсов. Шардов всегда не хватит. Со первой схемой шардинга вы в любом случае промахнетесь — всегда придется либо увеличивать количество серверов докидывать, либо что-то еще делать — в общем, как-то данные переукладывать.

Хорошо, если у нас приятные степени двойки: было 16 шардов-серверов, стало 32. Веселее, если было 17, стало 23 — два вазимно простых числа. Как же это делают базы данных, может быть, у них есть какая-то магия внутри?

Правильный ответ: нет, никакой магии внутри нет, у них внутри есть ад.

Дальше рассмотрим, что можно сделать «руками», авось поймем «как автоматом».

В лоб #1. Переселить всё

Для всех объектов считаем NewF(object), перекладываем на новый шард.

Вероятность совпадения NewF()=OldF() невелика.

Переложим почти всё вообще.

Ой.

Такого ада, как переложить все 2 млрд записей со старых шардов на новые, я надеюсь, нет нигде. Наивный подход понятен: было 17 машин, добавили 6 машин в кластер, перебрали 2 млрд записей, переложили их из 17 машин на 23 машины. Раз в 10 лет можно, наверное, даже и это сделать. Но в целом это плохой ход.

В лоб #2. Переселить половину

Следующее наивное улучшение — давайте откажемся от такой дурацкой схемы — запретим 17 машин решардить в 23, и всегда будем решардить 16 машин в 32 машины! Тогда нам по теории придется переложить ровно половину данных, и на практике мы тоже сможем это сделать.

Для всех объектов считаем NewF(object), перекладываем на новый шард.

Было строго 2^N, стало строго 2^(N+1) шардов.

Вероятность совпадения NewF()=OldF() равна 0,5.

Переложим примерно 50% данных.

Оптимально, но работает только для степеней двойки.

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

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

Хорошо, но неужели человечество не изобрело ничего больше — возникает вопрос у пытливого ума.

Веселее #3. Consistent hashing

Конечно, здесь обязательна картинка с кругом про consistent hashing.

Если загуглить «consistent hashing», то обязательно вылезет круг, вся выдача кругами заселена.

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

При добавлении шарда: просматриваем не всё, а только 2 «соседей», перекладываем в среднем 1/n.

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

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

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

Я рассказываю про этот космос с кругами, потому что, например, внутри Cassandra именно такая схема. То есть, когда она у вас начала записи между нодами гонять, знайте, что круг смотрит на вас и, наверное, не одобряет.

Тем не менее, по сравнению с первыми способами жизнь улучшилась — мы уже просматриваем при добавлении/удалении шарда не все записи, а только часть, и перекладываем только часть.

Внимание, вопрос: нельзя ли улучшить еще? И еще улучшить и равномерность загрузки шардов? — Говорят, что можно!

Веселее #4. Rendezvous/HRW

Следующая простая идея (материал же обучающий, поэтому ничего сложного): shard_id = arg max hash(object_id, shard_id).

Почему она называется Rendezvous hashing, я не знаю, но знаю, почему она называется Highest Random Weight. Ее очень просто визуализировать следующим образом:

У нас есть, например, 16 шардов. Для каждого объекта (строки), который надо куда-то положить, вычисляем 16 хэшей, зависящих от объекта с номера шарда. У кого самое высокое значение хэш-функции, тот и победил.

Это так называемый HRW-hashing, он же Rendezvous hashing. Тупая как палка схема вычисления номера шарда, во-первых, на глаз проще кругов и дает равномерную загрузку, с другой стороны.

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

Еще одна проблема, это вычислительно тяжело при большом количестве шардов.

Веселее #5. Еще техники

Интересно, что исследования не стоят на месте и Google каждый год публикует какую-нибудь новую космическую технику:

Jump Hash — Google ‘2014.
Multi Probe —Google ‘2015.
Maglev — Google ‘2016.

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

Выводы

Есть важная базовая техника под названием шардинг имени еще Галлия Юлия Цезаря: «Разделяй и властвуй, властвуй и разделяй!». Если данные не влазят в один сервер, надо их разбить на 20 серверов.

Узнав это все, должно возникнуть впечатление, что лучше бы не шардить. Если вы решили, что лучше бы не шардить — это правильное ощущение. Если можно добавить за 100$ памяти в сервер и ничего не шардить, то так и надо сделать. При шардировании появится сложная распределенная система с перекачкой данных туда-сюда, укладкой данных неизвестно куда. Если можно этого избежать — нужно этого избежать.

Лучше это делать не руками, лучше, чтобы «база» (поиск, DFS, ...) сама умела шардировать. В любом случае, рано или поздно, highload наступит и как-то данные дробить придется. Не факт, что даже если база умеет делать это сама, вы не нарветесь на какие-то проблемы. Помните про алгоритмический фундаментализм — надо понимать, как все устроено внутри.

Настраивая шардирование первый раз аккуратно выбирайте F(), думайте про запросы, сеть, и т.п. Но готовьтесь, вероятно, выбирать надо будет 2 раза и хотя бы раз придется все переделать.