1. Як зашардити і вгальмувати в N разів?

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

  • Надіслати запити docs00...docs15 послідовно, а не паралельно.
  • У простих запитах зробити вибірку не за ключем, WHERE something=234.

У цьому випадку серіалізована частина (serial) займає не 1% і не 5%, а приблизно 20% у сучасних базах даних. Можна отримати і 50% серіалізованої частини, якщо звертатися до бази даних з дико ефективного бінарного протоколу або лінкувати її як динамічну бібліотеку скриптом на Python.

Весь час обробки простого запиту займатимуть не розпаралелювані операції розбору запиту, підготовки плану тощо. Тобто гальмує не читання запису.

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

Раптово, при шардуванні важливий вибір мови програмування.

Пам'ятаємо про вибір мови програмування, тому що якщо надсилати запити до бази даних (або пошукового сервера) послідовно, то звідки взяти прискорення? Радше з'явиться уповільнення.

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), є зовнішні наліпки.

Конфігурування кожної конкретної наліпки — це окрема гігантська тема, яка не влізе в одну доповідь, тому ми обговоримо лише базові концепції. Давай краще поговоримо трохи про теорію кайфу.

3. Абсолютна ідеальна автоматика?

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

Шардування — це взагалі про що? У нас є 2 мільярди записів (або 64). Ми хочемо їх подрібнити на кілька шматків. Виникає несподіване питання: як? За яким принципом я свої 2 мільярди записів (або 64) повинен розкидати на доступні мені 16 серверів?

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

Вебінар JavaRush

Якщо заглибитися далі всередину математики, ця функція завжди залежить не тільки від самого об'єкта (самого рядка), але ще від зовнішніх налаштувань типу загальної кількості шардів. Функція, яка для кожного об'єкта повинна сказати, куди його класти, не може повертати значення більше, ніж є серверів системі. І функції трохи різні: 

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

Але далі ми не будемо занурюватися в ці нетрі окремих функцій: просто поговоримо які є чарівні функції F().

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% запитів стосуються останнього тижня, вони всі потраплятимуть на один шард, який цей останній тиждень обслуговує. Цей шард розплавиться, тоді як усі інші в цей час простоюватимуть. Водночас, викидати їх не можна, архівні дані зберігати також потрібно.

Не сказати, що це погана схема шардування: ми відрізали гарячі дані, проте з найбільш гарячим шардом треба щось робити.

Проблема вирішується ужимками, стрибками та припарками, тобто підвищенням кількості реплік для поточного дня, що горить. Потім — поступовим зниженням кількості реплік, коли цей день стає минулим і переходить до архіву. Тут немає ідеального рішення під назвою "треба просто чарівною хеш-функцією розмазати дані по кластеру іншим чином".

5. Ціна, яку потрібно заплатити

Формально ми тепер знаємо «все». Щоправда, ми не знаємо одного гігантського головного болю і двох головних болів менше.

1. Простий біль: погано розмазало

Це приклад із підручника, який у бою майже не зустрічається, але раптом.

  • Як приклад із датою, тільки без дати!
  • Ненавмисний нерівномірний (відчутно) розподіл.

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

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

Це проста проблема, чесно кажучи: я не думаю, що хоча б одна людина зі ста нарветься на це в житті, але раптом хоч комусь допоможе.

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

Як робити вибірки, які джойнять мільярд записів з однієї таблиці на мільярд записів з іншої таблиці?

  • Як «швидко» порахувати… WHERE randcol BETWEEN aaa AND bbb?
  • Як «спритно» зробити… users_32shards JOIN posts_1024 shards?

Коротка відповідь: ніяк, страждати!

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

Важливий момент — якісь запити завжди невдало розмажуться і будуть гальмувати. Важливо спробувати мінімізувати їхній відсоток. Як наслідок, не треба робити гігантські джойни із мільярдом на мільярд записів. Якщо є можливість маленьку таблицю, щодо якої джойниш у гігантській розшарденній таблиці, реплікувати на всі вузли, треба зробити це. Якщо джойни насправді локальні якимось чином, наприклад, є можливість користувача та його пости розмістити поруч, зашардити їх однаково, і всі джойни робити в межах однієї машини треба робити саме так.

Це окремий курс лекцій на три дні, тому переходимо до останнього пекельного болю та до різних алгоритмів боротьби з ним.

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 рази, і хоча б раз доведеться все перероблювати.