Apache HBase polega na zapewnieniu losowego dostępu do Big Data w czasie rzeczywistym z możliwością odczytu/zapisu, ale przede wszystkim w jaki sposób skutecznie wprowadzić te dane do HBase? Intuicyjnie nowy użytkownik spróbuje to zrobić za pośrednictwem interfejsów API klienta lub przy użyciu zadania MapReduce z TableOutputFormat, ale te podejścia są problematyczne, o czym dowiesz się poniżej. Zamiast tego funkcja ładowania zbiorczego HBase jest znacznie łatwiejsza w użyciu i może szybciej wstawić tę samą ilość danych.
Ten wpis na blogu przedstawi podstawowe koncepcje funkcji ładowania zbiorczego, przedstawi dwa przypadki użycia i zaproponuje dwa przykłady.
Przegląd ładowania zbiorczego
Jeśli masz którykolwiek z tych objawów, ładowanie zbiorcze jest prawdopodobnie właściwym wyborem dla Ciebie:
- Musisz dostosować swoje MemStore, aby wykorzystywały większość pamięci.
- Musisz albo użyć większych warstw WAL, albo całkowicie je ominąć.
- Twoje zagęszczanie i kolejki spłukiwania są w setkach.
- Twój GC jest poza kontrolą, ponieważ twoje wstawki mieszczą się w MB.
- Twoje opóźnienie wyczerpuje się z umowy SLA podczas importowania danych.
Większość z tych objawów jest powszechnie określana jako „bóle wzrostowe”. Korzystanie z ładowania zbiorczego może pomóc w ich uniknięciu.
W HBase-speak ładowanie zbiorcze to proces przygotowywania i ładowania HFiles (własny format pliku HBase) bezpośrednio do RegionServers, w ten sposób omijając ścieżkę zapisu i całkowicie omijając te problemy. Ten proces jest podobny do ETL i wygląda tak:
1. Wyodrębnij dane ze źródła, zwykle plików tekstowych lub innej bazy danych. HBase nie zarządza tą częścią procesu. Innymi słowy, nie możesz powiedzieć HBase, aby przygotował HFiles przez bezpośrednie odczytanie ich z MySQL — raczej musisz to zrobić własnymi środkami. Na przykład, możesz uruchomić mysqldump na stole i przesłać wynikowe pliki do HDFS lub po prostu pobrać pliki dziennika Apache HTTP. W każdym razie Twoje dane muszą być w HDFS przed następnym krokiem.
2. Przekształć dane w HFiles. Ten krok wymaga zadania MapReduce, a dla większości typów danych wejściowych będziesz musiał samodzielnie napisać Mapper. Zadanie będzie musiało wyemitować klucz wiersza jako Key i KeyValue, Put lub Delete jako wartość. Reduktor jest obsługiwany przez HBase; konfigurujesz go za pomocą HFileOutputFormat.configureIncrementalLoad() i wykonuje on następujące czynności:
- Sprawdza tabelę w celu skonfigurowania całkowitego partycjonowania zamówień
- Przesyła plik partycji do klastra i dodaje go do DistributedCache
- Ustawia liczbę zadań redukcji, aby odpowiadała bieżącej liczbie regionów
- Ustawia klasę klucza/wartości wyjściowej zgodnie z wymaganiami HFileOutputFormat
- Ustawia reduktor w celu wykonania odpowiedniego sortowania (albo KeyValueSortReducer albo PutSortReducer)
Na tym etapie w folderze wyjściowym zostanie utworzony jeden plik HFile na region. Pamiętaj, że dane wejściowe są prawie całkowicie przepisywane, więc będziesz potrzebować co najmniej dwa razy więcej miejsca na dysku niż rozmiar oryginalnego zestawu danych. Na przykład dla mysqldump 100GB powinieneś mieć co najmniej 200GB wolnego miejsca na dysku w HDFS. Możesz usunąć plik zrzutu na końcu procesu.
3. Załaduj pliki do HBase, informując RegionServers, gdzie je znaleźć. To najłatwiejszy krok. Wymaga użycia LoadIncrementalHFiles (bardziej znanego jako narzędzie completebulkload), a przekazując mu adres URL, który lokalizuje pliki w HDFS, załaduje każdy plik do odpowiedniego regionu za pośrednictwem serwera RegionServer, który go obsługuje. W przypadku podziału regionu po utworzeniu plików, narzędzie automatycznie podzieli plik HFile zgodnie z nowymi granicami. Ten proces nie jest zbyt wydajny, więc jeśli Twoja tabela jest aktualnie zapisywana przez inne procesy, najlepiej załadować pliki zaraz po zakończeniu etapu transformacji.
Oto ilustracja tego procesu. Przepływ danych przechodzi z oryginalnego źródła do HDFS, gdzie Serwery Regionów po prostu przeniosą pliki do katalogów swoich regionów.
Przypadki użycia
Wczytywanie oryginalnego zbioru danych: Wszyscy użytkownicy migrujący z innego magazynu danych powinni rozważyć ten przypadek użycia. Najpierw musisz wykonać ćwiczenie projektowania schematu tabeli, a następnie utworzyć samą tabelę, wstępnie podzieloną. Punkty podziału muszą uwzględniać rozkład klucza wiersza i liczbę serwerów regionalnych. Polecam przeczytanie prezentacji mojego kolegi Larsa George'a na temat zaawansowanego projektowania schematów dla każdego poważnego przypadku użycia.
Zaletą jest to, że zapisywanie plików jest znacznie szybsze niż przechodzenie przez ścieżkę zapisu serwera RegionServer (zapis zarówno do MemStore, jak i WAL), a następnie opróżnianie, kompaktowanie i tak dalej. Oznacza to również, że nie musisz dostrajać klastra do dużego obciążenia zapisu, a następnie ponownie dostrajać go do normalnego obciążenia.
Obciążenie przyrostowe: Załóżmy, że masz obecnie jakiś zestaw danych obsługiwany przez HBase, ale teraz musisz zaimportować więcej danych w partii od strony trzeciej lub masz nocne zadanie, które generuje kilka gigabajtów, które musisz wstawić. Prawdopodobnie nie jest tak duży, jak zbiór danych, który już obsługuje HBase, ale może wpłynąć na 95. percentyl opóźnienia. Przejście przez normalną ścieżkę zapisu będzie miało negatywny wpływ na wyzwolenie większej liczby opróżnień i zagęszczeń podczas importu niż normalnie. Ten dodatkowy stres we/wy będzie konkurował z zapytaniami wrażliwymi na opóźnienia.
Przykłady
Możesz użyć następujących przykładów we własnym klastrze Hadoop, ale instrukcje są dostępne dla maszyny wirtualnej Cloudera QuickStart, która jest klastrem z jednym węzłem, systemem operacyjnym gościa oraz przykładowymi danymi i przykładami wypiekanymi w urządzeniu maszyny wirtualnej dla Twojego komputera.
Po uruchomieniu maszyny wirtualnej powiedz jej za pomocą interfejsu internetowego, który otworzy się automatycznie, aby wdrożyć CDH, a następnie upewnij się, że usługa HBase jest również uruchomiona.
Wbudowana masowa ładowarka TSV
HBase jest dostarczany z zadaniem MR, które może odczytywać plik wartości rozdzielanych ogranicznikami i wyprowadzać dane bezpośrednio do tabeli HBase lub tworzyć pliki HFiles do ładowania zbiorczego. Tutaj zamierzamy:
- Pobierz przykładowe dane i prześlij je do HDFS.
- Uruchom zadanie ImportTsv, aby przekształcić plik w wiele plików HFiles zgodnie ze wstępnie skonfigurowaną tabelą.
- Przygotuj i załaduj pliki w HBase.
Pierwszym krokiem jest otwarcie konsoli i użycie następującego polecenia, aby pobrać przykładowe dane:
curl -O https://people.apache.org/~jdcryans/word_count.csv
Stworzyłem ten plik, uruchamiając licznik słów w oryginalnym rękopisie tego posta na blogu, a następnie wyświetlając wynik w formacie csv, bez żadnych tytułów kolumn. Teraz prześlij plik do HDFS:
hdfs dfs -put word_count.csv
Część pobierania zbiorczego ładowania jest teraz zakończona, musisz przekształcić plik. Najpierw musisz zaprojektować stół. Aby uprościć sprawę, nazwij to „liczba słów” — klawiszami wierszy będą same słowa, a jedyna kolumna będzie zawierać liczbę, w rodzinie, którą nazwiemy „f”. Najlepszą praktyką podczas tworzenia tabeli jest podzielenie jej zgodnie z rozkładem klawiszy wiersza, ale w tym przykładzie utworzymy tylko pięć regionów z punktami podziału rozmieszczonymi równomiernie w przestrzeni kluczy. Otwórz powłokę hbase:
hbase shell
I uruchom następujące polecenie, aby utworzyć tabelę:
create 'wordcount', {NAME => 'f'}, {SPLITS => ['g', 'm', 'r', 'w']} Cztery punkty podziału wygenerują pięć regionów, przy czym pierwszy region zaczyna się od pustego klucza wiersza. Aby uzyskać lepsze punkty podziału, możesz również przeprowadzić szybką analizę, aby zobaczyć, jak naprawdę rozkładają się słowa, ale zostawię to Tobie.
Jeśli skierujesz przeglądarkę swojej maszyny wirtualnej na http://localhost:60010/, zobaczysz naszą nowo utworzoną tabelę i jej pięć regionów, wszystkie przypisane do serwera RegionServer.
Teraz nadszedł czas na ciężkie podnoszenie. Wywołanie jara HBase w wierszu poleceń za pomocą skryptu „hadoop” spowoduje wyświetlenie listy dostępnych narzędzi. Ten, którego szukamy, nazywa się importtsv i ma następujące zastosowanie:
hadoop jar /usr/lib/hbase/hbase-0.94.6-cdh4.3.0-security.jar importtsv ERROR: Wrong number of arguments: 0 Usage: importtsv -Dimporttsv.columns=a,b,c
Linia poleceń, której będziemy używać, jest następująca:
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
Oto przegląd różnych elementów konfiguracji:
- -Dimporttsv.separator=, określa, że separatorem jest przecinek.
- -Dimporttsv.bulk.output=wyjście jest ścieżką względną do miejsca, w którym zostaną zapisane pliki HFiles. Ponieważ twój użytkownik na maszynie wirtualnej to domyślnie „cloudera”, oznacza to, że pliki będą znajdować się w /user/cloudera/output. Pominięcie tej opcji spowoduje, że zadanie będzie pisać bezpośrednio do HBase.
- -Dimporttsv.columns=HBASE_ROW_KEY,f:count to lista wszystkich kolumn zawartych w tym pliku. Klucz wiersza należy zidentyfikować za pomocą ciągu znaków HBASE_ROW_KEY z wielkimi literami; w przeciwnym razie nie rozpocznie pracy. (Zdecydowałem się użyć kwalifikatora „liczba”, ale może to być cokolwiek innego.)
Zadanie powinno zakończyć się w ciągu minuty, biorąc pod uwagę niewielki rozmiar danych wejściowych. Zwróć uwagę, że działa pięć Reduktorów, po jednym na region. Oto wynik na HDFS:
-rw-r--r-- 3 cloudera cloudera 4265 2013-09-12 13:13 output/f/2c0724e0c8054b70bce11342dc91897b -rw-r--r-- 3 cloudera cloudera 3163 2013-09-12 13:14 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 2013-09-12 13:14 output/f/c656d893bd704260a613be62bddb4d5f
Jak widać, pliki aktualnie należą do użytkownika „cloudera”. Aby je załadować, musimy zmienić właściciela na „hbase” lub HBase nie będzie miał uprawnień do przenoszenia plików. Uruchom następujące polecenie:
sudo -u hdfs hdfs dfs -chown -R hbase:hbase/user/cloudera/output
W ostatnim kroku musimy użyć narzędzia completebulkload, aby wskazać, gdzie znajdują się pliki i do których tabel ładujemy:
hbase org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles output wordcount
Wracając do powłoki HBase, możesz uruchomić polecenie count, które pokaże ci, ile wierszy zostało załadowanych. Jeśli zapomniałeś chown, polecenie się zawiesi.
Niestandardowe zadanie MR
Masowy moduł ładujący TSV jest dobry do prototypowania, ale ponieważ interpretuje wszystko jako ciągi i nie obsługuje manipulowania polami w czasie transformacji, będziesz musiał napisać własne zadanie MR. Mój kolega James Kinley, który pracuje jako architekt rozwiązań w Europie, napisał taką pracę, którą wykorzystamy w naszym następnym przykładzie. Dane dotyczące tego zadania zawierają publiczne wiadomości na Facebooku i Twitterze dotyczące finałów NBA 2010 (gra 1) między Lakers i Celtics. Kod znajdziesz tutaj. (Maszyna wirtualna Quick Start jest dostarczana z zainstalowanymi git i maven, dzięki czemu możesz sklonować na niej repozytorium.)
Patrząc na klasę Driver, najważniejsze bity są następujące:
job.setMapOutputKeyClass(ImmutableBytesWritable.class);
job.setMapOutputValueClass(KeyValue.class);
…
// Auto configure partitioner and reducer
HFileOutputFormat.configureIncrementalLoad(job, hTable);
Najpierw program Mapper musi wyprowadzić ImmutableBytesWritable, który zawiera klucz wiersza, a wartością wyjściową może być KeyValue, Put lub Delete. Drugi fragment pokazuje, jak skonfigurować Reduktor; w rzeczywistości jest całkowicie obsługiwany przez HFileOutputFormat. configureIncrementalLoad() zgodnie z opisem we wcześniejszej części „Transformacja”.
Klasa HBaseKVMapper zawiera tylko Mapper, który respektuje skonfigurowany klucz wyjściowy i wartości:
public class HBaseKVMapper extends
Mapper<LongWritable, Text, ImmutableBytesWritable,
KeyValue> { Aby go uruchomić, musisz skompilować projekt za pomocą maven i pobrać pliki danych, korzystając z łączy w README. (Zawiera również skrypt powłoki do tworzenia tabeli.) Przed rozpoczęciem zadania nie zapomnij przesłać plików do HDFS i ustawić ścieżkę klasy, aby być świadomym HBase, ponieważ tym razem nie będziesz używać jego jar :
export HADOOP_CLASSPATH=$HADOOP_CLASSPATH:/etc/hbase/conf/:/usr/lib/hbase/*
Będziesz mógł rozpocząć zadanie za pomocą wiersza poleceń podobnego do tego:
hadoop jar hbase-examples-0.0.1-SNAPSHOT.jar com.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.jar RowFeeder\ for\ Celtics\ and\ Lakers\ Game\ 1.csv output2 NBAFinal2010
Jak widać, zależności pracy należy dodać osobno. Na koniec możesz załadować pliki, najpierw zmieniając ich właściciela, a następnie uruchamiając narzędzie do kompletnego ładowania zbiorczego:
sudo -u hdfs hdfs dfs -chown -R hbase:hbase/user/cloudera/output2 hbase org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles output2 NBAFinal2010
Potencjalne problemy
Ponownie pojawiają się ostatnio usunięte dane. Ten problem występuje, gdy polecenie Delete jest wstawiane przez ładowanie zbiorcze i jest mocno skompaktowane, podczas gdy odpowiedni Put nadal znajduje się w MemStore. Dane zostaną uznane za usunięte, gdy Delete znajduje się w HFile, ale gdy zostaną usunięte podczas kompaktowania, Put stanie się ponownie widoczne. Jeśli masz taki przypadek użycia, rozważ skonfigurowanie rodzin kolumn, aby zachować usunięte komórki za pomocą KEEP_DELETED_CELLS w powłoce lub HColumnDescriptor.setKeepDeletedCells().
Danych ładowanych zbiorczo nie można zastąpić innym ładowaniem zbiorczym. Ten problem występuje, gdy dwa ładowane zbiorczo pliki HFile ładowane w różnym czasie próbują zapisać inną wartość w tej samej komórce, co oznacza, że mają ten sam klucz wiersza, rodzinę, kwalifikator i sygnaturę czasową. W rezultacie zostanie zwrócona pierwsza wstawiona wartość zamiast drugiej. Ten błąd zostanie naprawiony w HBase 0.96.0 i CDH 5 (następna główna wersja CDH), a prace są wykonywane w HBASE-8521 dla gałęzi 0.94 i CDH 4.
Ładowanie zbiorcze powoduje duże zagęszczenie. Ten problem pojawia się, gdy wykonujesz przyrostowe ładowanie zbiorcze, a liczba ładowanych zbiorczo plików jest wystarczająca, aby wywołać niewielkie kompaktowanie (domyślny próg to 3). Pliki HFiles są ładowane z numerem sekwencji ustawionym na 0, więc są pobierane jako pierwsze, gdy RegionServer wybiera pliki do kompaktowania, az powodu błędu wybierze również wszystkie pozostałe pliki. Ten problem poważnie dotknie tych, którzy mają już duże regiony (wiele GB) lub często ładują się zbiorczo (co kilka godzin i mniej), ponieważ duża ilość danych zostanie skompaktowana. HBase 0.96.0 ma odpowiednią poprawkę, podobnie jak CDH 5; HBASE-8521 rozwiązuje problem w 0.94, ponieważ ładowane masowo pliki HFiles mają teraz przypisany prawidłowy numer sekwencji. HBASE-8283 można włączyć za pomocą hbase.hstore.useExploringCompation po wersji 0.94.9 i CDH 4.4.0, aby złagodzić ten problem, będąc po prostu inteligentniejszym algorytmem selekcji kompaktowania.
Dane ładowane zbiorczo nie są replikowane . Ponieważ ładowanie zbiorcze omija ścieżkę zapisu, WAL nie jest zapisywany w ramach procesu. Replikacja działa poprzez odczytywanie plików WAL, więc nie zobaczy danych załadowanych zbiorczo — i to samo dotyczy edycji, które używają Put.setWriteToWAL(true). Jednym ze sposobów poradzenia sobie z tym jest przesłanie nieprzetworzonych plików lub plików HFiles do innego klastra i wykonanie tam innego przetwarzania.
Wniosek
Celem tego wpisu na blogu było przedstawienie podstawowych pojęć związanych z ładowaniem zbiorczym Apache HBase. Wyjaśniliśmy, jak ten proces przypomina wykonywanie ETL i że jest znacznie lepszy w przypadku zestawów danych big data niż przy użyciu normalnego interfejsu API, ponieważ omija ścieżkę zapisu. Dwa przykłady zostały dołączone, aby pokazać, jak proste pliki TSV można ładować zbiorczo do HBase i jak napisać własny program Mapper dla innych formatów danych.
Teraz możesz spróbować zrobić to samo, korzystając z graficznego interfejsu użytkownika za pośrednictwem Hue.
