Как да:Използвайте HBase групово зареждане и защо

Apache HBase има за цел да ви даде произволен достъп в реално време за четене/запис до вашите големи данни, но как ефективно да получите тези данни в HBase на първо място? Интуитивно, нов потребител ще се опита да направи това чрез клиентските API или чрез използване на задача MapReduce с TableOutputFormat, но тези подходи са проблематични, както ще научите по-долу. Вместо това, функцията за групово зареждане на HBase е много по-лесна за използване и може да вмъкне същото количество данни по-бързо.

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

Общ преглед на груповото зареждане

Ако имате някой от тези симптоми, груповото натоварване вероятно е правилният избор за вас:

Трябваше да настроите MemStores, за да използвате по-голямата част от паметта.
Трябваше или да използвате по-големи WAL, или да ги заобиколите изцяло.
Вашите опашки за уплътняване и измиване са стотици.
Вашият GC е извън контрол, защото вашите вмъквания варират в MB.
Вашето забавяне излиза извън SLA, когато импортирате данни.

Повечето от тези симптоми обикновено се наричат „нарастващи болки“. Използването на групово зареждане може да ви помогне да ги избегнете.

В HBase-speak груповото зареждане е процесът на подготовка и зареждане на HFiles (собствен файлов формат на HBase) директно в RegionServers, като по този начин се заобикаля пътя за запис и се елиминират напълно тези проблеми. Този процес е подобен на ETL и изглежда така:

1. Извличане на данните от източник, обикновено текстови файлове или друга база данни. HBase не управлява тази част от процеса. С други думи, не можете да кажете на HBase да подготви HFiles, като ги чете директно от MySQL — по-скоро трябва да го направите със собствени средства. Например, можете да стартирате mysqldump на таблица и да качите получените файлове в HDFS или просто да вземете вашите Apache HTTP регистрационни файлове. Във всеки случай данните ви трябва да са в HDFS преди следващата стъпка.

2. Преобразувайте данните в HFiles. Тази стъпка изисква задание MapReduce и за повечето типове вход ще трябва сами да напишете Mapper. Задачата ще трябва да издаде ключа на реда като ключ и или KeyValue, Put или Delete като стойност. Редукторът се управлява от HBase; конфигурирате го с помощта на HFileOutputFormat.configureIncrementalLoad() и той прави следното:

Проверява таблицата, за да конфигурира разделител за обща поръчка
Качва файла на дяловете в клъстера и го добавя към DistributedCache
Задава броя на задачите за намаляване, за да съответства на текущия брой региони
Задава класа на изходния ключ/стойност, за да съответства на изискванията на HFileOutputFormat
Настройва редуктора за извършване на подходящото сортиране (или KeyValueSortReducer или PutSortReducer)

На този етап ще бъде създаден по един HFile за регион в изходната папка. Имайте предвид, че входните данни са почти напълно пренаписани, така че ще ви трябва поне два пъти повече налично дисково пространство от размера на оригиналния набор от данни. Например, за 100GB mysqldump трябва да имате поне 200GB налично дисково пространство в HDFS. Можете да изтриете дъмп файла в края на процеса.

3. Заредете файловете в HBase, като кажете на RegionServers къде да ги намерят. Това е най-лесната стъпка. Той изисква използване на LoadIncrementalHFiles (по-известен като инструмент за пълно зареждане) и като му предаде URL, който намира файловете в HDFS, той ще зареди всеки файл в съответния регион чрез RegionServer, който го обслужва. В случай, че регионът е бил разделен след създаването на файловете, инструментът автоматично ще раздели HFile според новите граници. Този процес не е много ефективен, така че ако вашата таблица в момента се записва от други процеси, най-добре е файловете да бъдат заредени веднага щом стъпката на трансформиране бъде извършена.

Ето илюстрация на този процес. Потокът от данни преминава от оригиналния източник към HDFS, където регионалните сървъри просто ще преместят файловете в директориите на своите региони.

Случаи на употреба

Оригинално зареждане на набор от данни: Всички потребители, които мигрират от друго хранилище за данни, трябва да обмислят този случай на употреба. Първо, трябва да преминете през упражнението по проектиране на схемата на таблицата и след това да създадете самата таблица, предварително разделена. Точките на разделяне трябва да вземат предвид разпределението на ключовете на редове и броя на регионалните сървъри. Препоръчвам да прочетете презентацията на моя колега Ларс Джордж относно усъвършенстваното проектиране на схеми за всеки сериозен случай на употреба.

Предимството тук е, че е много по-бързо да се запишат файловете директно, отколкото да се премине през пътя за запис на RegionServer (запис както в MemStore, така и в WAL) и след това евентуално да се промиват, уплътняват и т.н. Това също така означава, че не е нужно да настройвате клъстера си за натоварване с тежък запис и след това да го настройвате отново за нормалното си натоварване.

Постепенно натоварване: Да кажем, че имате някакъв набор от данни, който в момента се обслужва от HBase, но сега трябва да импортирате повече данни в пакет от трета страна или имате нощна работа, която генерира няколко гигабайта, които трябва да вмъкнете. Вероятно не е толкова голям, колкото набора от данни, който HBase вече обслужва, но може да повлияе на 95-ия процентил на вашата латентност. Преминаването през нормалния път на запис ще има неблагоприятен ефект от задействане на повече изтривания и уплътнения по време на импортирането от нормалното. Този допълнителен стрес за IO ще се конкурира с вашите чувствителни към забавяне заявки.

Примери

Можете да използвате следните примери във вашия собствен клъстер Hadoop, но инструкциите са предоставени за Cloudera QuickStart VM, която е клъстер с един възел, гост ОС и примерни данни и примери, изпечени във виртуална машина за вашия работен плот.

След като стартирате VM, кажете му чрез уеб интерфейса, който автоматично ще се отвори, да разгърне CDH и след това се уверете, че услугата HBase също е стартирана.

Вграден насипен товарач TSV

HBase се доставя с MR задание, което може да чете файл със стойности, разделени с разделители, и да изведе директно в таблица на HBase или да създава HFiles за групово зареждане. Ето ние ще:

Вземете примерните данни и ги качете в HDFS.
Изпълнете заданието ImportTsv, за да трансформирате файла в множество HFiles според предварително конфигурирана таблица.
Подгответе и заредете файловете в HBase.

Първата стъпка е да отворите конзола и да използвате следната команда, за да получите примерни данни:

curl -Ohttps://people.apache.org/~jdcryans/word_count.csv

Създадох този файл, като изпълних преброяване на думи върху оригиналния ръкопис на тази публикация в блога и след това изведех резултата във формат csv, без никакви заглавия на колони. Сега качете файла в HDFS:

hdfs dfs -поставете word_count.csv

Извличането на частта от насипното натоварване сега е завършено, трябва да трансформирате файла. Първо трябва да проектирате масата. За да запазите нещата прости, наречете го „wordcount“ — ключовете на редовете ще бъдат самите думи и единствената колона ще съдържа броя, в семейство, което ще наречем „f“. Най-добрата практика при създаване на таблица е да я разделите според разпределението на ключовете на редове, но за този пример просто ще създадем пет региона с точки на разделяне, разпределени равномерно в ключовото пространство. Отворете обвивката hbase:

hbase shell

И изпълнете следната команда, за да създадете таблицата:

създайте 'wordcount', {NAME => 'f'}, {SPLITS => ['g', 'm', 'r', 'w']}

Четирите точки на разделяне ще генерират пет региона, където първият регион започва с празен ключ за ред. За да получите по-добри точки на разделяне, можете също да направите бърз анализ, за да видите как думите наистина са разпределени, но ще оставя това на вас.

Ако насочите браузъра на вашата VM към http://localhost:60010/, ще видите нашата новосъздадена таблица и нейните пет региона, всички присвоени на RegionServer.

Сега е време да извършите тежкото вдигане. Извикването на HBase jar в командния ред със скрипта „hadoop“ ще покаже списък с налични инструменти. Този, който искаме, се нарича importtsv и има следната употреба:

hadoop jar /usr/lib/hbase/hbase-0.94.6-cdh4.3.0-security.jar importtsv ГРЕШКА:Грешен брой аргументи:0 Използване:importtsv -Dimporttsv.columns=a,b,c

Командният ред, който ще използваме, е следният:

hadoop jar /usr/lib/hbase/hbase-0.94.6-cdh4.3.0-security.jar importtsv-Dimporttsv.separator=,-Dimporttsv.bulk.output=output-Dimporttsv.columns=HBASE_ROW_KEY,f:wordcount word_count.csv

Ето кратко описание на различните конфигурационни елементи:

-Dimporttsv.separator=, указва, че разделителят е запетая.
-Dimporttsv.bulk.output=output е относителен път до мястото, където ще бъдат записани HFiles. Тъй като вашият потребител на VM е „cloudera“ по подразбиране, това означава, че файловете ще бъдат в /user/cloudera/output. Пропускането на тази опция ще накара заданието да пише директно в HBase.
-Dimporttsv.columns=HBASE_ROW_KEY,f:count е списък на всички колони, съдържащи се в този файл. Ключът на реда трябва да бъде идентифициран с помощта на низа HBASE_ROW_KEY с всички главни букви; в противен случай няма да започне работата. (Реших да използвам квалификатора „count“, но може да е нещо друго.)

Работата трябва да приключи в рамките на една минута, като се има предвид малкият размер на входа. Имайте предвид, че работят пет редуктора, по един за регион. Ето резултата на HDFS:

-rw-r--r-- 3 cloudera cloudera 4265 2013-09-12 13:13 output/f/2c0724e0c8054b70bce11342dc91897b-rw-r--3 cloudera 1 3 cloudera 1 3 cloudera 1 6 3 3 cloudera 1 ера Изход/F/786198CA47AE406F9BE05C9EB09BEB36-RW-R-R-3 CLOUDERA CLOUDERA 2487 2013-09-12 13:14 Изход/F/9B0E5B2A137E479CBC978132E3FC84D2-RW-R-R-R-R-3 CLOUDER CLOUDER 29612-RW-R-R-R-R-R-ROUDER CLouder 296 :13 output/f/bb341f04c6d845e8bb95830e9946a914-rw-r--r-- 3 cloudera cloudera 1336 2013-09-12 13:14 output/f/c656d7db20d7d0b20d7db

Както можете да видите, файловете в момента принадлежат на потребителя „cloudera“. За да ги заредим, трябва да променим собственика на „hbase“ или HBase няма да има разрешение да премести файловете. Изпълнете следната команда:

sudo -u hdfs hdfs dfs -chown -R hbase:hbase/user/cloudera/output

За последната стъпка трябва да използваме инструмента за пълно зареждане, за да посочим къде са файловете и кои таблици зареждаме:

hbase org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles изходен брой думи

Връщайки се в обвивката на HBase, можете да изпълните командата count, която ще ви покаже колко реда са били заредени. Ако сте забравили да chown, командата ще увисне.

Персонализирана MR работа

TSV зареждачът за масово натоварване е добър за създаване на прототипи, но тъй като интерпретира всичко като низове и не поддържа манипулиране на полетата по време на трансформация, в крайна сметка ще трябва да напишете своя собствена MR работа. Моят колега Джеймс Кинли, който работи като архитект на решения в Европа, написа такава работа, която ще използваме за следващия си пример. Данните за работата съдържат публични съобщения във Facebook и Twitter, свързани с финалите на НБА през 2010 г. (мач 1) между Лейкърс и Селтикс. Можете да намерите кода тук. (Виртуалната машина за бърз старт идва с инсталирани git и maven, за да можете да клонирате хранилището в нея.)

Разглеждайки класа Driver, най-важните битове са следните:

job.setMapOutputKeyClass(ImmutableBytesWritable.class); job.setMapOutputValueClass(KeyValue.class);… // Автоматично конфигуриране на разделителя и редуктор HFileOutputFormat.configureIncrementalLoad(job, hTable);

Първо, вашият Mapper трябва да изведе ImmutableBytesWritable, който съдържа ключа на реда, а изходната стойност може да бъде KeyValue, Put или Delete. Вторият фрагмент показва как да конфигурирате Reducer; всъщност се обработва напълно от HFileOutputFormat. configureIncrementalLoad(), както е описано в секцията „Трансформиране“ по-рано.

Класът HBaseKVMapper съдържа само Mapper, който спазва конфигурирания изходен ключ и стойности:

public class HBaseKVMapper разширява Mapper {

За да го стартирате, ще трябва да компилирате проекта с помощта на maven и да вземете файловете с данни, следвайки връзките в README. (Той също така съдържа скрипта на обвивката за създаване на таблицата.) Преди да започнете заданието, не забравяйте да качите файловете в HDFS и да настроите своя път към класа да е наясно с HBase, защото този път няма да използвате неговия jar :

експортиране HADOOP_CLASSPATH=$HADOOP_CLASSPATH:/etc/hbase/conf/:/usr/lib/hbase/*

Ще можете да стартирате заданието с помощта на команден ред, подобен на този:

hadoop jar hbase-examples-0.0.1-SNAPSHOT.jarcom.cloudera.examples.hbase.bulkimport.Driver -libjars/home/cloudera/.m2/repository/joda-time/joda-time/2.1/joda- time-2.1.jar,/home/cloudera/.m2/repository/net/sf/opencsv/opencsv/2.3/opencsv-2.3.jarRowFeeder\ for\ Celtics\ and\ Lakers\ Game\ 1.csv output2 NBAFinal2010

Както можете да видите, зависимостите на заданието трябва да се добавят отделно. И накрая, можете да заредите файловете, като първо промените собственика им и след това стартирате инструмента за пълно зареждане:

sudo -u hdfs hdfs dfs -chown -R hbase:hbase/user/cloudera/output2hbase org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles output2 NBAFinal2010

Потенциални проблеми

Наскоро изтритите данни се появяват отново. Този проблем се случва, когато Delete се вмъкне чрез насипно натоварване и е силно уплътнено, докато съответното Put все още е в MemStore. Данните ще се считат за изтрити, когато Delete е в HFile, но след като бъде премахнат по време на уплътняването, Put ще стане видим отново. Ако имате такъв случай на употреба, помислете дали да не конфигурирате семействата си колони, за да запазите изтритите клетки с KEEP_DELETED_CELLS в обвивката или HColumnDescriptor.setKeepDeletedCells().

Групово заредените данни не могат да бъдат презаписани от друго групово зареждане. Този проблем възниква, когато два групово заредени HFiles, заредени в различно време, се опитват да напишат различна стойност в една и съща клетка, което означава, че имат един и същ ключ на ред, семейство, квалификатор и времева марка. Резултатът е, че първата вмъкната стойност ще бъде върната вместо втората. Тази грешка ще бъде коригирана в HBase 0.96.0 и CDH 5 (следващата основна версия на CDH) и се работи в HBASE-8521 за клон 0.94 и CDH 4.

Насипното натоварване задейства големи уплътнения. Този проблем възниква, когато правите нарастващи групови зареждания и има достатъчно файлове с групово зареждане, за да предизвикат незначително уплътняване (прагът по подразбиране е 3). HFiles се зареждат с пореден номер, зададен на 0, така че те се качват първи, когато RegionServer избира файлове за уплътняване и поради грешка ще избере и всички останали файлове. Този проблем ще засегне сериозно тези, които вече имат големи региони (много GB) или които зареждат често (на всеки няколко часа и по-малко), тъй като много данни ще бъдат уплътнени. HBase 0.96.0 има правилната корекция, както и CDH 5; HBASE-8521 коригира проблема в 0.94, тъй като на групово заредените HFiles вече е присвоен правилен пореден номер. HBASE-8283 може да бъде активиран с hbase.hstore.useExploringCompation след 0.94.9 и CDH 4.4.0, за да смекчи този проблем, като просто е по-интелигентен алгоритъм за избор на уплътняване.

Групово заредените данни не се репликират . Тъй като груповото зареждане заобикаля пътя за запис, WAL не се записва като част от процеса. Репликацията работи чрез четене на WAL файловете, така че да не вижда групово заредените данни – и същото важи за редакциите, които използват Put.setWriteToWAL(true). Един от начините да се справите с това е да изпратите необработените файлове или HFiles към другия клъстер и да извършите другата обработка там.

Заключение

Целта на тази публикация в блога беше да ви запознае с основните концепции за групово зареждане на Apache HBase. Обяснихме как процесът е като правенето на ETL и че е много по-добре за големи набори от данни, отколкото да използвате нормалния API, тъй като заобикаля пътя за запис. Двата примера бяха включени, за да покажат как прости TSV файлове могат да бъдат групово заредени в HBase и как да напишете свой собствен Mapper за други формати на данни.

Сега можете да опитате да направите същото, като използвате графичен потребителски интерфейс чрез Hue.