Så här gör du:Använd HBase Bulk Loading och varför

Apache HBase handlar om att ge dig slumpmässig läs-/skrivåtkomst i realtid till din Big Data, men hur får du effektivt in den datan i HBase i första hand? Intuitivt kommer en ny användare att försöka göra det via klientens API:er eller genom att använda ett MapReduce-jobb med TableOutputFormat, men dessa tillvägagångssätt är problematiska, som du kommer att lära dig nedan. Istället är HBase-bulkladdningsfunktionen mycket enklare att använda och kan infoga samma mängd data snabbare.

Det här blogginlägget kommer att introducera de grundläggande koncepten för bulkladdningsfunktionen, presentera två användningsfall och föreslå två exempel.

Översikt över bulkladdning

Om du har något av dessa symtom är bulkladdning förmodligen rätt val för dig:

Du behövde finjustera dina MemStores för att använda det mesta av minnet.
Du behövde antingen använda större WALs eller förbigå dem helt.
Dina packnings- och spolköer är i hundratal.
Din GC är utom kontroll eftersom dina infogningsintervall i MBs.
Din latens försvinner från din SLA när du importerar data.

De flesta av dessa symtom kallas vanligtvis för "växtvärk". Att använda bulkladdning kan hjälpa dig att undvika dem.

I HBase-speak är bulkladdning processen att förbereda och ladda hFiles (HBases eget filformat) direkt till RegionServers, och på så sätt kringgå skrivvägen och eliminera dessa problem helt. Denna process liknar ETL och ser ut så här:

1. Extrahera data från en källa, vanligtvis textfiler eller en annan databas. HBase hanterar inte denna del av processen. Med andra ord, du kan inte säga åt HBase att förbereda hFiles genom att direkt läsa dem från MySQL – snarare måste du göra det på egen hand. Du kan till exempel köra mysqldump på en tabell och ladda upp de resulterande filerna till HDFS eller bara ta dina Apache HTTP-loggfiler. Din data måste i alla fall finnas i HDFS innan nästa steg.

2. Omvandla data till hFiles. Detta steg kräver ett MapReduce-jobb och för de flesta inmatningstyper måste du skriva Mapper själv. Jobbet kommer att behöva avge radnyckeln som nyckel, och antingen ett nyckelvärde, ett sätt eller ett radera som värde. Reduceraren hanteras av HBase; du konfigurerar det med HFileOutputFormat.configureIncrementalLoad() och det gör följande:

Inspekterar tabellen för att konfigurera en total orderpartitionerare
Laddar upp partitionsfilen till klustret och lägger till den i DistributedCache
Ställer in antalet reduceringsuppgifter för att matcha det aktuella antalet regioner
Ställer in utdatanyckel/värdeklass för att matcha HFileOutputFormats krav
Ställer in reduceraren för att utföra lämplig sortering (antingen KeyValueSortReducer eller PutSortReducer)

I detta skede kommer en HF-fil att skapas per region i utdatamappen. Tänk på att indata är nästan helt omskrivna, så du kommer att behöva minst dubbelt så mycket tillgängligt diskutrymme än storleken på den ursprungliga datamängden. Till exempel, för en 100 GB mysqldump bör du ha minst 200 GB tillgängligt diskutrymme i HDFS. Du kan ta bort dumpfilen i slutet av processen.

3. Ladda filerna i HBase genom att tala om för RegionServers var de kan hitta dem. Detta är det enklaste steget. Det kräver att du använder LoadIncrementalHFiles (mer allmänt känt som completebulkload-verktyget), och genom att skicka det en URL som lokaliserar filerna i HDFS, kommer det att ladda varje fil till den relevanta regionen via RegionServer som betjänar den. I händelse av att en region delades efter att filerna skapades kommer verktyget automatiskt att dela upp HF-filen enligt de nya gränserna. Den här processen är inte särskilt effektiv, så om din tabell för närvarande skrivs till av andra processer är det bäst att ladda filerna så snart omvandlingssteget är klart.

Här är en illustration av denna process. Dataflödet går från den ursprungliga källan till HDFS, där RegionServers helt enkelt flyttar filerna till sina regioners kataloger.

Användningsfall

Original datauppsättning: Alla användare som migrerar från en annan databutik bör överväga detta användningsfall. Först måste du gå igenom övningen med att designa tabellschemat och sedan skapa själva tabellen, fördelad. Splitpunkterna måste ta hänsyn till radnyckelfördelningen och antalet RegionServers. Jag rekommenderar att du läser min kollega Lars Georges presentation om avancerad schemadesign för alla seriösa användningsfall.

Fördelen här är att det är mycket snabbare att skriva filerna direkt än att gå igenom RegionServers skrivväg (att skriva till både MemStore och WAL) och sedan så småningom spola, komprimera och så vidare. Det betyder också att du inte behöver ställa in ditt kluster för en skrivtung arbetsbelastning och sedan ställa in det igen för din normala arbetsbelastning.

Inkrementell belastning: Låt oss säga att du har en datauppsättning som för närvarande betjänas av HBase, men nu måste du importera mer data i batch från en tredje part eller så har du ett nattjobb som genererar några gigabyte som du behöver infoga. Den är förmodligen inte lika stor som den datauppsättning som HBase redan visar, men den kan påverka din latenss 95:e percentil. Att gå igenom den normala skrivvägen kommer att ha den negativa effekten att utlösa fler spolningar och packningar under importen än normalt. Denna ytterligare IO-stress kommer att konkurrera med dina latenskänsliga frågor.

Exempel

Du kan använda följande exempel i ditt eget Hadoop-kluster men instruktionerna tillhandahålls för Cloudera QuickStart VM, som är ett ennodskluster, gäst-OS och exempeldata och exempel inbakade i en virtuell maskinapparat för ditt skrivbord.

När du startar den virtuella datorn, berätta för den, via webbgränssnittet som öppnas automatiskt, att distribuera CDH och se sedan till att HBase-tjänsten också startas.

Inbyggd TSV Bulk Loader

HBase levereras med ett MR-jobb som kan läsa en avgränsningsseparerad värdefil och mata ut direkt i en HBase-tabell eller skapa HFiles för bulkladdning. Här ska vi:

Hämta exempeldata och ladda upp den till HDFS.
Kör ImportTsv-jobbet för att omvandla filen till flera HFiles enligt en förkonfigurerad tabell.
Förbered och ladda filerna i HBase.

Det första steget är att öppna en konsol och använda följande kommando för att hämta exempeldata:

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

Jag skapade den här filen genom att köra en ordräkning på originalmanuskriptet till just detta blogginlägg och sedan mata ut resultatet i csv-format, utan några kolumntitlar. Ladda nu upp filen till HDFS:

hdfs dfs -put word_count.csv

När utvinningsdelen av bulkladdningen nu är klar måste du transformera filen. Först måste du designa bordet. För att göra det enkelt, kalla det "ordräkning" - radnycklarna kommer att vara själva orden och den enda kolumnen kommer att innehålla räkningen, i en familj som vi kallar "f". Den bästa metoden när du skapar en tabell är att dela upp den enligt radnyckelfördelningen, men för det här exemplet skapar vi bara fem regioner med delade punkter jämnt fördelade över nyckelutrymmet. Öppna hbase-skalet:

hbase-skal

Och kör följande kommando för att skapa tabellen:

skapa 'wordcount', {NAME => 'f'}, {SPLITS => ['g', 'm', 'r', 'w']}

De fyra delade punkterna kommer att generera fem regioner, där den första regionen börjar med en tom radnyckel. För att få bättre delade poäng kan du också göra en snabb analys för att se hur orden verkligen är fördelade, men jag överlåter det till dig.

Om du pekar på din virtuella dators webbläsare till http://localhost:60010/ kommer du att se vår nyskapade tabell och dess fem regioner alla tilldelade RegionServer.

Nu är det dags att göra det tunga arbetet. Om du anropar HBase-burken på kommandoraden med "hadoop"-skriptet visas en lista över tillgängliga verktyg. Den vi vill ha heter importtsv och har följande användning:

hadoop jar /usr/lib/hbase/hbase-0.94.6-cdh4.3.0-security.jar importtsv FEL:Fel antal argument:0 Användning:importtsv -Dimporttsv.columns=a,b,c

Kommandoraden vi ska använda är följande:

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:count wordcount word_count.csv

Här är en sammanfattning av de olika konfigurationselementen:

-Dimporttsv.separator=, anger att avgränsaren är ett kommatecken.
-Dimporttsv.bulk.output=output är en relativ sökväg till var HFiles kommer att skrivas. Eftersom din användare på den virtuella datorn är "cloudera" som standard betyder det att filerna kommer att vara i /user/cloudera/output. Om du hoppar över det här alternativet kommer jobbet att skrivas direkt till HBase.
-Dimporttsv.columns=HBASE_ROW_KEY,f:count är en lista över alla kolumner som finns i den här filen. Radnyckeln måste identifieras med strängen HBASE_ROW_KEY med stora bokstäver; annars startar det inte jobbet. (Jag bestämde mig för att använda kvalet "räkna", men det kan vara något annat.)

Jobbet bör slutföras inom en minut, med tanke på den lilla inmatningsstorleken. Observera att fem Reducers körs, en per region. Här är resultatet på HDFS:

-rw-r--r-- 3 cloudera cloudera 4265 2013-09-12 13:13 output/f/2c0724e0c8054b70bce11342dc91897b-rw-er- 4265 2013-09-12 13:13 output/f/2c0724e0c8054b70bce11342dc91897b-rw-era--r-- 1 output/f/786198ca47ae406f9be05c9eb09beb36-rw-r--r-- 3 cloudera cloudera 2487 2013-09-12 13:14 output/f/9b0e5b2a137e479cbc978132e3fc84d2-rw-r--r-- 3 cloudera cloudera 2961 2013-09-12 13 :13 output/f/bb341f04c6d845e8bb95830e9946a914-rw-r--r-- 3 cloudera cloudera 1336 1336 2013-09-12 13:14 output/f/c656badd8423bd6423bd6423bd6456d7423bd6456d7423bdd6456d849bdd>

Som du kan se tillhör filerna för närvarande användaren "cloudera". För att ladda dem måste vi ändra ägaren till "hbase" annars har HBase inte behörighet att flytta filerna. Kör följande kommando:

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

För det sista steget måste vi använda completebulkload-verktyget för att peka på var filerna finns och vilka tabeller vi laddar till:

hbase org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles output wordcount

När du går tillbaka till HBase-skalet kan du köra count-kommandot som visar hur många rader som laddades. Om du glömde att chown kommer kommandot att hänga sig.

Anpassat MR-jobb

TSV-bulklastaren är bra för prototyper, men eftersom den tolkar allt som strängar och inte stöder manipulering av fälten vid transformationstid, kommer du att behöva skriva ditt eget MR-jobb. Min kollega James Kinley, som arbetar som lösningsarkitekt i Europa, skrev ett sådant jobb som vi kommer att använda för vårt nästa exempel. Uppgifterna för jobbet innehåller offentliga Facebook- och Twitter-meddelanden relaterade till 2010 års NBA-final (match 1) mellan Lakers och Celtics. Du hittar koden här. (Snabbstarts-VM kommer med git och maven installerade så att du kan klona förvaret på den.)

Om man tittar på Driver-klassen är de viktigaste bitarna följande:

job.setMapOutputKeyClass(ImmutableBytesWritable.class); job.setMapOutputValueClass(KeyValue.class);... // Autokonfigurera partitionerare och reducerare HFileOutputFormat.configureIncrementalLoad(job, hTable);

Först måste din Mapper mata ut en ImmutableBytesWritable som innehåller radnyckeln, och utdatavärdet kan vara antingen ett KeyValue, ett Put eller en Delete. Det andra utdraget visar hur man konfigurerar Reducer; det hanteras faktiskt helt av HFileOutputFormat. configureIncrementalLoad() enligt beskrivningen i avsnittet "Transformera" tidigare.

Klassen HBaseKVMapper innehåller bara Mapper som respekterar den konfigurerade utdatanyckeln och värden:

public class HBaseKVMapper utökar Mapper {

För att köra det måste du kompilera projektet med hjälp av maven och ta tag i datafilerna genom att följa länkarna i README. (Den innehåller även skalskriptet för att skapa tabellen.) Glöm inte att ladda upp filerna till HDFS och ställa in din klasssökväg för att vara medveten om HBase innan du startar jobbet, eftersom du inte kommer att använda dess jar :

exportera HADOOP_CLASSPATH=$HADOOP_CLASSPATH:/etc/hbase/conf/:/usr/lib/hbase/*

Du kommer att kunna starta jobbet med en kommandorad som liknar denna:

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\ och\ Lakers\ Game\ 1.csv output2 NBAFinal2010

Som du kan se måste jobbets beroenden läggas till separat. Slutligen kan du ladda filerna genom att först byta ägare och sedan köra komplett bulkload-verktyget:

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

Möjliga problem

Nyligen raderad data dyker upp igen. Det här problemet inträffar när en Delete infogas via en bulklast och är kraftigt komprimerad medan motsvarande Put fortfarande finns i en MemStore. Uppgifterna kommer att anses vara raderade när raderingen finns i en HF-fil, men när den väl har tagits bort under komprimeringen kommer Putten att bli synlig igen. Om du har ett sådant användningsfall, överväg att konfigurera dina kolumnfamiljer för att behålla de raderade cellerna med KEEP_DELETED_CELLS i skalet eller HColumnDescriptor.setKeepDeletedCells().

Massladdad data kan inte skrivas över av en annan massinläsning. Det här problemet uppstår när två bulkladdade HFiler som laddas vid olika tidpunkter försöker skriva ett annat värde i samma cell, vilket innebär att de har samma radnyckel, familj, kvalificerare och tidsstämpel. Resultatet är att det första infogade värdet kommer att returneras istället för det andra. Denna bugg kommer att fixas i HBase 0.96.0 och CDH 5 (nästa CDH-huvudversion) och arbete pågår i HBASE-8521 för 0.94-grenen och CDH 4.

Bulklastning utlöser stora packningar. Det här problemet uppstår när du gör inkrementella bulklaster och det finns tillräckligt med bulkladdade filer för att utlösa en mindre komprimering (standardtröskeln är 3). HFilerna laddas med ett sekvensnummer satt till 0 så att de hämtas först när RegionServer väljer filer för en komprimering, och på grund av en bugg kommer den också att välja alla återstående filer. Det här problemet kommer allvarligt att påverka de som redan har stora regioner (flera GB) eller som masslastar ofta (med några timmars mellanrum och mindre) eftersom mycket data kommer att komprimeras. HBase 0.96.0 har rätt fix och så kommer CDH 5; HBASE-8521 åtgärdar problemet i 0.94 eftersom de bulkladdade HF-filerna nu tilldelas ett korrekt sekvensnummer. HBASE-8283 kan aktiveras med hbase.hstore.useExploringCompation efter 0.94.9 och CDH 4.4.0 för att lindra detta problem genom att bara vara en smartare algoritm för val av kompaktering.

Massinlästa data replikeras inte . Eftersom massladdning går förbi skrivvägen skrivs inte WAL till som en del av processen. Replikering fungerar genom att läsa WAL-filerna så att den inte ser massinlästa data – och detsamma gäller för redigeringarna som använder Put.setWriteToWAL(true). Ett sätt att hantera det är att skicka råfilerna eller HFiles till det andra klustret och göra den andra bearbetningen där.

Slutsats

Målet med det här blogginlägget var att introducera dig till Apache HBase bulk loadings grundläggande koncept. Vi förklarade hur processen är som att göra ETL, och att det är mycket bättre för stora datamängder än att använda det vanliga API:et eftersom det går förbi skrivvägen. De två exemplen inkluderades för att visa hur enkla TSV-filer kan massladdas till HBase och hur du skriver din egen Mapper för andra dataformat.

Nu kan du prova att göra samma sak med ett grafiskt användargränssnitt via Hue.