Теория и практика на използването на 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 следи сърдечния ритъм на активния на узел ZooKeeper и при изчезване поема функциите на мастера.

2. Записване на данни в HBASE

Първо ще разгледаме най-простия случай – запис на обект ключ-стойност в определена таблица с помощта на put(rowkey). Клиентът първо трябва да установи къде се намира коренният регионален сървър (Root Region Server — RRS), който съхранява таблицата 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, възстановяваме времето за обработка до нормалното.

Освен увеличението на 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 000 повторения на своята операция.
• Размерът на записа в блока беше 100 байта, 1000 байта или 10000 байта (случайно).
• Блоковете се стартираха с различен брой запитвани ключове (или един ключ, или 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