Теорія та практика використання HBase

Добридень! Мене звуть Данило Ліповий, наша команда в Сбертеху почала використовувати HBase як сховище оперативних даних. У ході його вивчення накопичився досвід, який захотілося систематизувати та описати (сподіваємося, що багатьом буде корисно). Всі наведені нижче експерименти проводилися з версіями HBase 1.2.0-cdh5.14.2 та 2.0.0-cdh6.0.0-beta1.

1. Загальна архітектура
2. Запис даних у HBASE
3. Читання даних з HBASE
4. Кешування даних
5. Пакетна обробка даних MultiGet/MultiPut
6. Стратегія розбивки таблиць на регіони
7. Відмовостійкість, компактифікація та локальність даних
8. Налаштування та продуктивність
9. Тестування навантаження
10. Висновки

1. Загальна архітектура

Резервний Master слухає heartbeat активного на вузлі ZooKeeper і у разі зникнення бере функції майстра на себе.

2. Запис даних у HBASE

Спочатку розглянемо найпростіший випадок – запис об'єкта ключ-значение на якусь таблицю з допомогою put(rowkey). Клієнт спочатку повинен з'ясувати, де розташований кореневий регіон сервер (RORS), який зберігає таблицю hbase:meta. Цю інформацію він одержує від ZooKeeper. Після чого він звертається до RRS і читає таблицю hbase:meta, з якої витягує інформацію, який RegionServer (RS) відповідає за зберігання даних по заданому ключі rowkey в таблиці, що цікавить його. З метою подальшого використання мета-таблиця кешується клієнтом і тому наступні звернення йдуть швидше безпосередньо до RS.

Далі RS, отримавши запит, перш за все пише його в WriteAheadLog (WAL), що необхідно для відновлення у разі падіння. Потім зберігає дані в MemStore. Це буфер у пам'яті, який містить відсортований набір ключів регіону. Таблиця може бути розбита на регіони (партиції), кожен з яких містить набір ключів, що не перетинається. Це дозволяє, розмістивши регіони різних серверах, отримувати вищу продуктивність. Однак, незважаючи на очевидність цього твердження, ми побачимо, що це працює не у всіх випадках.

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

При виконанні операції Delete фізичного видалення даних не відбувається. Вони просто позначаються як віддалені, а саме знищення відбувається у момент виклику функції major compact, яку детальніше написано в п.7.

Файли у форматі HFile накопичуються в HDFS і іноді запускається процес minor compact, який просто склеює маленькі файли в більші, нічого не видаляючи. Згодом це перетворюється на проблему, яка проявляється тільки при читанні даних (до цього повернемося трохи пізніше).

Крім описаного вище процесу завантаження є набагато ефективніша процедура, в якій полягає мабуть найсильніша сторона цієї БД - BulkLoad. Вона полягає в тому, що ми самостійно формуємо HFiles і підкладаємо на диск, що дозволяє чудово масштабуватися і досягати пристойних швидкостей. По суті, обмеженням тут не HBase, а можливості заліза. Нижче наведено результати завантаження на кластері, що складається з 16 RegionServers та 16 NodeManager YARN (CPU Xeon E5-2680 v4 @ 2.40GHz * 64 потоку), версія HBase 1.2.0-cdh5.14.2.

Тут видно, що збільшуючи кількість партицій (регіонів) в таблиці, а також екзекуторів Spark, отримуємо збільшення швидкості завантаження. Також швидкість залежить від обсягу запису. Великі блоки дають приріст у вимірі МБ/сек, дрібні кількості вставлених записів в одиницю часу, при інших рівних.

Також можна запустити завантаження в дві таблиці одночасно та отримати подвоєння швидкості. Нижче видно, що запис блоків 10 КБ відразу дві таблиці йде зі швидкістю близько 600 Мб/сек в кожну (сумарно 1275 Мб/сек), що збігається зі швидкістю запису в одну таблицю 623 МБ/сек (див. №11 вище)

А ось другий запуск із записами в 50 КБ показує, що швидкість завантаження зростає вже незначно, що говорить про наближення до граничних значень. При цьому потрібно мати на увазі, що на HBASE тут навантаження практично не створюється, все що від нього потрібно, це спочатку віддати дані з hbase:meta, а після підкладки HFiles, скинути дані BlockCache і зберегти буфер MemStore на диск, якщо він не порожній.

3. Читання даних з HBASE

Якщо вважати, що вся інформація з hbase:meta вже має клієнта (див. п.2), то запит йде відразу на той RS, де зберігається потрібний ключ. Спочатку пошук здійснюється в MemCache. Незалежно від того, є дані чи ні, пошук здійснюється також у буфері BlockCache і при необхідності в HFiles. Якщо дані були знайдені у файлі, вони поміщаються в BlockCache і при наступному запиті будуть повернуті швидше. Пошук HFile відбувається відносно швидко завдяки використанню фільтра Блюма, тобто. Вважаючи невеликий обсяг даних, він відразу визначає, чи містить цей файл потрібний ключ і якщо ні, то переходить до наступного.

Отримавши дані із цих трьох джерел RS формує відповідь. Зокрема, він може передати відразу кілька знайдених версій об'єкта, якщо клієнт запросив версійність.

4. Кешування даних

Буфери MemStore і BlockCache займають до 80% виділеної on-heap пам'яті RS (інше зарезервовано для сервісних завдань RS). Якщо типовий режим використання такий, що процеси пишуть і одразу читають ці дані, то має сенс зменшити BlockCache і збільшити MemStore, т.к. При записі дані в кеш на читання не потрапляють, використання BlockCache буде відбуватися рідше. Буфер BlockCache складається з двох частин: LruBlockCache (завжди on-heap) та BucketCache (як правило off-heap або SSD). BucketCache варто використовувати, коли запитів читання дуже багато і вони не містяться в LruBlockCache, що призводить до активної роботи Garbage Collector. При цьому радикального зростання продуктивності від використання кешу на читання чекати не варто, проте до цього ми ще повернемося до п. 8

BlockCache один на весь RS, а MemStore для кожної таблиці свій (по одному на кожен Column Family).

Як описано в теорії, при записі дані в кеш не потрапляють і дійсно такі параметри CACHE_DATA_ON_WRITE для таблиці і «Cache DATA on Write» для RS встановлені в false. Однак на практиці, якщо записати дані в MemStore, потім скинути його на диск (очистивши таким чином), потім видалити файл, що вийшов, то виконавши get запит ми успішно отримаємо дані. Причому навіть якщо зовсім відключити BlockCache і забити таблицю новими даними, потім домогтися скидання MemStore на диск, видалити їх і запросити з іншої сесії, вони все одно звідки витягнуться. Отже HBase зберігає у собі як дані, а й таємничі загадки.

hbase(main):001:0> create 'ns:magic', 'cf'
Created table ns:magic
Took 1.1533 seconds
hbase(main):002:0> put 'ns:magic', 'key1', 'cf:c', 'try_to_delete_me'
Took 0.2610 seconds
hbase(main):003:0> flush 'ns:magic'
Took 0.6161 seconds
hdfs dfs -mv /data/hbase/data/ns/magic/* /tmp/trash
hbase(main):002:0> get 'ns:magic', 'key1'
 cf:c      timestamp=1534440690218, value=try_to_delete_me

Параметр Cache DATA on Read встановлено false. Якщо є ідеї, ласкаво просимо обговорити це у коментарях.

5. Пакетна обробка даних MultiGet/MultiPut

Обробка одиночних запитів (Get/Put/Delete) досить дорога операція, тому слід об'єднувати наскільки можна їх у List чи List, що дозволяє отримувати значний приріст продуктивності. Особливо це стосується операції запису, а при читанні є наступний підводний камінь. Нижче показано час читання 50 000 записів з MemStore. Читання проводилося в один потік і горизонтальною осі показано кількість ключів у запиті. Тут видно, що зі збільшенням тисячі ключів щодо одного запиту час виконання падає, тобто. швидкість збільшується. Однак при включеному за умовчанням режимі MSLAB після цього порога починається радикальне падіння продуктивності, причому чим більше обсяг даних запису, тим більше часу роботи.

Тести виконували на віртуалці, 8 ядер, версія HBase 2.0.0-cdh6.0.0-beta1.

Режим MSLAB має зменшити фрагментацію heap, яка виникає через перемішування даних нового та старого поколінь. Як вирішення проблеми при включенні MSLAB дані поміщаються у відносно невеликі осередки (chunk) та обробляються порціями. Через війну, коли обсяг у запрошеному пакеті даних перевищує виділений розмір, продуктивність різко падає. З іншого боку, вимкнення даного режиму також не бажане, оскільки призведе до зупинок через GC в моменти інтенсивної роботи з даними. Хорошим виходом є збільшення обсягів осередку, у разі активного запису через put одночасно з читанням. Варто відзначити, що проблема не виникає якщо після запису виконувати команду flush, яка скидає MemStore на диск або якщо здійснюється завантаження за допомогою BulkLoad. У таблиці нижче показано, що запити MemStore даних більшого обсягу (і однакової кількості) призводять до уповільнення. Однак збільшуючи chunksize повертаємо час обробки до норми.

p align="justify"> Крім збільшення chunksize допомагає дроблення даних по регіонах, тобто. сплітінг таблиць. Це призводить до того, що на кожен регіон приходить менша кількість запитів і якщо вони розміщуються в осередку, відгук залишається добрим.

6. Стратегія розбиття таблиць на регіони (спілітинг)

Так як HBase є key-value сховищем і партиціонування здійснюється за ключом, то дуже важливо розділяти дані рівномірно по всіх регіонах. Наприклад, партиціонування такої таблиці на три частини призведе до того, що дані будуть розбиті на три регіони:

Буває, що це призводить до різкого уповільнення, якщо завантажувані надалі дані будуть мати вигляд наприклад long значень в більшості своїй починаються з однієї і тієї ж цифри, наприклад:

1000001
1000002
...
1100003

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

HexStringSplit – Перетворює ключ у рядок із шістнадцятковим кодуванням у діапазоні «00000000» => «FFFFFFFF» і заповнюючи зліва нулями.

UniformSplit – Перетворює ключ на масив байт із шістнадцятковим кодуванням у діапазоні «00» => «FF» і заповнюючи праворуч нулями.

Крім того, можна вказати будь-який діапазон або набір ключів для розбиття і налаштувати автосплітінг. Однак одним з найбільш простих та ефективних підходів є UniformSplit та використання конкатенації хеша, наприклад старшої пари байт від прогону ключа через функцію CRC32(rowkey) та власне rowkey:

hash + rowkey

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

7. Відмовостійкість та локальність даних

Оскільки за кожен набір ключів відповідає лише один регіон, вирішенням проблем, пов'язаних з падіннями RS або виведенням з експлуатації, є зберігання всіх необхідних даних у HDFS. При падінні RS майстер виявляє це через відсутність heartbeat на вузлі ZooKeeper. Тоді він призначає обслуговуваний регіон іншому RS і так як HFiles зберігаються в розподіленій файловій системі, новий господар вичитує їх і продовжує обслуговувати дані. Однак, оскільки частина даних може бути в MemStore і не встигла потрапити в HFiles, для відновлення історії операцій використовується WAL, які також зберігаються у HDFS. Після накату змін, RS здатний відповідати запити, проте переїзд призводить до того, частина даних і процеси їх обслуговуючі опиняються на різних нодах, тобто. знижується місцевість.

Вирішенням проблеми є major compaction – ця процедура переміщає файли на ті ноди, які за них відповідають (туди де розташовані їхні регіони), внаслідок чого під час цієї процедури різко зростає навантаження на мережу та диски. Проте доступ до даних помітно прискорюється. Крім того, major_compaction виконує об'єднання всіх HFiles в один файл у межах регіону, а також очищає дані залежно від налаштувань таблиці. Наприклад, можна задати кількість версій об'єкта, яке необхідно зберігати або час його життя, після якого об'єкт фізично видаляється.

Ця процедура може справити позитивний вплив на роботу HBase. На зображенні нижче видно, як деградувала продуктивність в результаті активного запису даних. Тут видно як одну таблицю 40 потоків писали і 40 потоків одночасно читали дані. Пишучі потоки формують дедалі більше HFiles, які вираховуються іншими потоками. В результаті все більше даних потрібно видаляти з пам'яті і, зрештою, починає працювати GC, який практично паралізує всю роботу. Запуск major compaction привів до чищення завалів, що утворилися, і відновлення продуктивності.

Тест виконувався на 3-х DataNode та 4-х RS (CPU Xeon E5-2680 v4 @ 2.40GHz * 64 потоки). Версія HBase 1.2.0-cdh5.14.2

Варто відзначити, що запуск major compaction виконувався на «живій» таблиці, в яку активно писали та читали дані. У мережі зустрічалося твердження, що це може призвести до некоректної відповіді під час читання даних. Для перевірки був запущений процес, який генерував нові дані та писав їх у таблицю. Після чого відразу ж читав і звіряв чи збігається отримане значення з тим, що було записано. Під час роботи цього процесу близько 200 разів запускався major compaction і жодного збою не зафіксовано. Можливо, проблема проявляється рідко і тільки під час високого завантаження, тому безпечніше все-таки планово зупиняти процеси запису та читання і виконувати очищення не допускаючи таких просадок GC.

Також major compaction не впливає на стан MemStore, для скидання його на диск та компактифікації потрібно використовувати flush (connection.getAdmin().flush(TableName.valueOf(tblName))).

8. Налаштування та продуктивність

Як вже було сказано, найбільший успіх HBase показує там, де йому нічого не потрібно робити при виконанні BulkLoad. Втім, це стосується більшості систем та людей. Однак цей інструмент підходить швидше для масового укладання даних великими блоками, тоді як якщо процес вимагає виконання безлічі конкуруючих запитів на читання і запис, використовуються описані вище команди Get і Put. Для визначення оптимальних параметрів було здійснено запуски при різних комбінаціях параметрів таблиць та налаштувань:

• Запускалося 10 потоків одночасно 3 рази поспіль (назвемо блоком потоків).
• Час роботи всіх потоків у блоці усереднювався і був підсумковим результатом роботи блоку.
• Усі потоки працювали з тією самою таблицею.
• Перед кожним запуском блоку потоків виконувався major compaction.
• Кожен блок виконував лише одну з таких операцій:

- Put
- Get
- Get + Put

• Кожен блок виконував 50 повторень своєї операції.
• Розмір запису в блоці 100 байт, 1000 байт чи 10000 байт (random).
• Блоки запускалися з різною кількістю запитуваних ключів (або один ключ або 10).
• Блоки запускалися при різних параметрах таблиці. Змінювалися параметри:

— BlockCache = вмикався або вимикався
- BlockSize = 65 Кб або 16 Кб
- Партій = 1, 5 або 30
— MSLAB = увімкнено або вимкнено

Таким чином блок виглядає так:

a. Вмикався/вимикався режим MSLAB.
b. Створювалася таблиця, на який встановлювалися такі параметри: BlockCache = true/none, BlockSize = 65/16 Kb, Партиций = 1/5/30.
c. Встановлювався стиск GZ.
d. Запускалося 10 потоків одночасно які роблять 1/10 операцій put/get/get+put у цю таблицю записами по 100/1000/10000 байт, виконуючи 50 000 запитів поспіль (ключи рандомні).
e. Пункт d повторювався тричі.
f. Час роботи всіх потоків усереднювався.

Було перевірено всі можливі комбінації. Передбачувано, що при збільшенні розміру запису швидкість падатиме або що вимкнення кешування призведе до уповільнення. Однак мета була зрозуміти ступінь та значущість впливу кожного параметра, тому зібрані дані були подані на вхід функції лінійної регресії, що дає змогу оцінити достовірність за допомогою статистики t. Нижче наведено результати роботи блоків виконують операції Put. Повний набір комбінацій 2 * 2 * 3 * 2 * 3 = 144 варіанти + 72 т.к. деякі були виконані двічі. Тож у сумі 216 запусків:

Тестування проводилося на міні-кластері що складається з 3-х DataNode і 4-х RS (CPU Xeon E5-2680 v4 @ 2.40GHz * 64 потоки). Версія HBase 1.2.0-cdh5.14.2.

Найбільш висока швидкість вставки 3.7 с була отримана при вимкненому режимі MSLAB, на таблиці з однією партицією, з включеним BlockCache, BlockSize = 16, записами по 100 байт по 10 штук в пачці.
Найбільш низька швидкість вставки 82.8 с була отримана при включеному режимі MSLAB, на таблиці з однією партицією, з включеним BlockCache, BlockSize = 16, записами по 10000 байт по 1 штуці.

Тепер подивимося на модель. Ми бачимо хорошу якість моделі по R2, але зрозуміло, що екстраполяція тут протипоказана. Реальна поведінка системи при зміні параметрів буде лінійною, ця модель потрібна не для прогнозів, а для розуміння, що сталося в межах заданих параметрів. Наприклад тут бачимо за критерієм Стьюдента, що з операції Put немає значення параметри BlockSize і BlockCache (що взагалі цілком передбачувано):

А ось те, що збільшення кількості партій веде до зниження продуктивності дещо несподівано (ми вже бачили позитивний вплив збільшення кількості партій при BulkLoad), хоча і зрозуміло. По-перше, для обробки доводиться формувати запити до 30 регіонів замість одного, а обсяг даних не такий, щоб це дало виграш. По-друге загальний час роботи визначається найповільнішим RS, оскільки кількість DataNode менше кількості RS частина регіонів мають нульову локальність. Ну і подивимося на п'ятірку лідерів:

Тепер оцінимо результати виконання блоків Get:

Кількість партицій втратила значимість, що ймовірно пояснюється тим, що дані добре кешуються і кеш для читання найбільш значущий (статистично) параметр. Природно, що збільшення кількості повідомлень у запиті - теж дуже корисно для продуктивності. Найкращі результати:

Ну і нарешті подивимося на модель блоку, який виконував спочатку get, а потім put:

Тут усі параметри значимі. І результати лідерів:

9. Навантажувальне тестування

Ну і нарешті запустимо більш-менш пристойне навантаження, але завжди цікавіше, коли є з чим порівнювати. На сайті DataStax – ключового розробника Cassandra є результати НТ ряду NoSQL сховищ, у тому числі HBase версії 0.98.6-1. Завантаження здійснювалось 40 потоками, розмір даних 100 байт, диски SSD. Результат тестування операцій Read-Modify-Write показав такі результати.

Наскільки я зрозумів, читання здійснювалося блоками по 100 записів і на 16 нод HBase тест DataStax показав продуктивність 10 тис. операцій на секунду.

Вдало, що в нашому кластері теж 16 нод, але не дуже «вдало», що на кожному по 64 ядра (потоку), тоді як у тесті DataStax лише по 4. З іншого боку, у них диски SSD, а у нас HDD і більше нова версія HBase та утилізація CPU під час навантаження практично збільшувалася не значно (візуально на 5-10 відсотків). Проте спробуємо запуститися цієї конфігурації. Налаштування таблиць за замовчуванням читання проводиться в діапазоні ключів від 0 до 50 млн. випадковим чином (тобто по суті щоразу новий). У таблиці 50 мільйонів записів розбито на 64 партиції. Ключі захешовані crc32. Налаштування таблиць дефолтні, MSLAB увімкнено. Запуск 40 потоків, кожен потік читає набір зі 100 випадкових ключів і відразу пише згенеровані 100 байт по цих ключах назад.

Стенд: 16 DataNode і 16 RS (CPU Xeon E5-2680 v4 @ 2.40GHz * 64 потоки). Версія HBase 1.2.0-cdh5.14.2.

Середній результат ближче до 40 тис. операцій на секунду, що значно краще, ніж у тесті DataStax. Проте з метою експерименту можна змінити умови. Досить малоймовірно, що вся робота вестиметься виключно з однією таблицею, а також з унікальними ключами. Припустимо, що є якийсь «гарячий» набір ключів, який генерує основне навантаження. Тому спробуємо створити навантаження більшими записами (10 КБ), також пачками по 100, в 4 різних таблиці і обмеживши діапазон запитуваних ключів 50 тис. На графіку нижче показаний запуск 40 потоків, кожен потік читає набір зі 100 ключів і тут же пише випадкові 10 КБ за цими ключами тому.

Стенд: 16 DataNode і 16 RS (CPU Xeon E5-2680 v4 @ 2.40GHz * 64 потоки). Версія HBase 1.2.0-cdh5.14.2.

У ході навантаження кілька разів запускався major compaction, як було показано вище без цієї процедури, продуктивність буде поступово деградувати, проте під час виконання також виникає додаткове навантаження. Просідання викликані різними причинами. Іноді потоки закінчували роботу і поки вони перезапускалися, виникала пауза, іноді сторонні програми створювали навантаження на кластер.

Читання і одразу ж запис — один із найважчих сценаріїв роботи для HBase. Якщо робити тільки put запити невеликого розміру, наприклад, по 100 байт, об'єднавши їх у пачки по 10-50 тис штук, можна отримати сотні тисяч операцій на секунду і аналогічно справи з запитами тільки на читання. Варто зазначити, що результати радикально кращі за ті, що вийшли у DataStax найбільше за рахунок запитів блоками по 50 тис.

Стенд: 16 DataNode і 16 RS (CPU Xeon E5-2680 v4 @ 2.40GHz * 64 потоки). Версія HBase 1.2.0-cdh5.14.2.

10. висновки

Ця система досить гнучко налаштовується, проте вплив великої кількості параметрів все ще залишається невідомим. Частина з них була протестована, але не увійшла до результуючого набору тестів. Наприклад, попередні експерименти показали незначну значимість такого параметра як DATA_BLOCK_ENCODING, який кодує інформацію, використовуючи значення із сусідніх осередків, що цілком зрозуміло для даних, що згенерували випадковим чином. У разі використання великої кількості об'єктів, що повторюються, виграш може бути значним. В цілому можна сказати, що HBase справляє враження досить серйозної та продуманої БД, яка при операціях з великими блоками даних може бути досить продуктивною. Особливо якщо є можливість рознести у часі процеси читання та запису.

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

Джерело: habr.com