Spark-on-HBase:złącze HBase oparte na DataFrame

Ten wpis na blogu został opublikowany na Hortonworks.com przed fuzją z Cloudera. Niektóre linki, zasoby lub odniesienia mogą nie być już dokładne.

Z dumą ogłaszamy techniczną wersję zapoznawczą złącza Spark-HBase, opracowanego przez Hortonworks we współpracy z Bloomberg.

Łącznik Spark-HBase wykorzystuje interfejs API źródła danych (SPARK-3247) wprowadzony w Spark-1.2.0. Wypełnia lukę między prostym magazynem HBase Key Value a złożonymi relacyjnymi zapytaniami SQL i umożliwia użytkownikom przeprowadzanie złożonej analizy danych w oparciu o HBase przy użyciu Spark. HBase DataFrame to standardowa ramka Spark DataFrame, która może współpracować z dowolnymi innymi źródłami danych, takimi jak Hive, ORC, Parquet, JSON itp.

Tło

Istnieje kilka łączników Spark HBase o otwartym kodzie źródłowym dostępnych jako pakiety Spark, jako niezależne projekty lub w trunku HBase.

Spark został przeniesiony do interfejsów API Dataset/DataFrame, które zapewniają wbudowaną optymalizację planu zapytań. Teraz użytkownicy końcowi wolą używać interfejsu opartego na DataFrames/Datasets.

Złącze HBase w trunku HBase posiada bogate wsparcie na poziomie RDD m.in. BulkPut itp., ale jego obsługa DataFrame nie jest tak bogata. Złącze magistrali HBase opiera się na standardowym HadoopRDD z wbudowanym formatem TableInputFormat HBase ma pewne ograniczenia wydajności. Dodatkowo BulkGet wykonywany w sterowniku może być pojedynczym punktem awarii.

Istnieje kilka innych alternatywnych implementacji. Weź Spark-SQL-on-HBase jako przykład. Stosuje bardzo zaawansowane niestandardowe techniki optymalizacji, osadzając własny plan optymalizacji zapytań w standardowym silniku Spark Catalyst, przesyła RDD do HBase i wykonuje skomplikowane zadania, takie jak częściowa agregacja, wewnątrz koprocesora HBase. Takie podejście jest w stanie osiągnąć wysoką wydajność, ale jest trudne do utrzymania ze względu na jego złożoność i szybką ewolucję Sparka. Również zezwolenie na uruchamianie dowolnego kodu wewnątrz koprocesora może stanowić zagrożenie dla bezpieczeństwa.

Złącze Spark-on-HBase (SHC) zostało opracowane w celu przezwyciężenia tych potencjalnych wąskich gardeł i słabości. Implementuje standardowy interfejs Spark Datasource API i wykorzystuje silnik Spark Catalyst do optymalizacji zapytań. Równolegle RDD jest konstruowany od podstaw zamiast używania TableInputFormat w celu osiągnięcia wysokiej wydajności. Dzięki temu dostosowanemu RDD można zastosować i w pełni wdrożyć wszystkie krytyczne techniki, takie jak przycinanie partycji, przycinanie kolumn, przekazywanie predykatów i lokalizowanie danych. Konstrukcja sprawia, że ​​konserwacja jest bardzo łatwa, a jednocześnie zapewnia dobry kompromis między wydajnością a prostotą.

Architektura

Zakładamy, że Spark i HBase są wdrożone w tym samym klastrze, a programy wykonawcze Spark są zlokalizowane razem z serwerami regionu, jak pokazano na poniższym rysunku.

Rysunek 1. Architektura złącza Spark-on-HBase

Na wysokim poziomie łącznik traktuje funkcje Scan i Get w podobny sposób, a obie akcje są wykonywane w executorach. Sterownik przetwarza zapytanie, agreguje skany/pobiera na podstawie metadanych regionu i generuje zadania według regionu. Zadania są wysyłane do preferowanych executorów znajdujących się w tym samym miejscu co serwer regionu i są wykonywane równolegle w executorach w celu uzyskania lepszej lokalizacji danych i współbieżności. Jeśli region nie przechowuje wymaganych danych, do tego serwera regionu nie przypisano żadnego zadania. Zadanie może składać się z wielu skanowań i pobierania zbiorczego, a żądania danych przez zadanie są pobierane tylko z jednego serwera regionalnego, a ten serwer regionalny będzie również preferowaną lokalizacją dla zadania. Należy zauważyć, że sterownik nie bierze udziału w rzeczywistym wykonywaniu zadań, z wyjątkiem zadań planowania. Dzięki temu kierowca nie jest wąskim gardłem.

Katalog tabel

Aby przenieść tabelę HBase jako relacyjną tabelę do platformy Spark, definiujemy mapowanie między tabelami HBase i Spark o nazwie Katalog tabel. Ten katalog składa się z dwóch krytycznych części. Jedna to definicja klucza wiersza, a druga to mapowanie między kolumną tabeli w Spark i rodziną kolumn i kwalifikatorem kolumn w HBase. Aby uzyskać szczegółowe informacje, zapoznaj się z sekcją Wykorzystanie.

Natywna obsługa Avro

Łącznik natywnie obsługuje format Avro, ponieważ bardzo powszechną praktyką jest utrwalanie uporządkowanych danych w HBase jako tablica bajtów. Użytkownik może utrwalić rekord Avro bezpośrednio w HBase. Wewnętrznie schemat Avro jest automatycznie konwertowany na natywny typ danych Spark Catalyst. Należy zauważyć, że obie części klucz-wartość w tabeli HBase można zdefiniować w formacie Avro. Proszę zapoznać się z przykładami/przypadkami testowymi w repozytorium w celu dokładnego użycia.

Predykat w dół

Łącznik pobiera tylko wymagane kolumny z serwera regionu, aby zmniejszyć obciążenie sieci i uniknąć nadmiarowego przetwarzania w aparacie Spark Catalyst. Istniejące standardowe filtry HBase są używane do wykonywania naciskania predykatów bez wykorzystywania możliwości koprocesora. Ponieważ HBase nie zna typu danych z wyjątkiem tablicy bajtów i niespójności kolejności między typami pierwotnymi Java a tablicą bajtów, musimy wstępnie przetworzyć warunek filtru przed ustawieniem filtru w operacji skanowania, aby uniknąć utraty danych. Na serwerze regionu rekordy niespełniające warunku zapytania są odfiltrowywane.

Przycinanie partycji

Wyodrębniając klucz wiersza z predykatów, dzielimy funkcję Scan/BulkGet na wiele nienakładających się zakresów, tylko serwery regionu, które mają żądane dane, wykonają funkcję Scan/BulkGet. Obecnie przycinanie partycji odbywa się na pierwszym wymiarze kluczy wierszy. Na przykład, jeśli klucz wiersza to „klucz1:klucz2:klucz3”, czyszczenie partycji będzie oparte tylko na „kluczu1”. Zwróć uwagę, że warunki WHERE muszą być dokładnie określone. W przeciwnym razie przycinanie partycji może nie zadziałać. Na przykład, WHERE rowkey1> „abc” OR kolumna =„xyz” (gdzie rowkey1 to pierwszy wymiar klucza wiersza, a kolumna to zwykła kolumna hbase) spowoduje pełne skanowanie, ponieważ musimy pokryć wszystkie zakresy, ponieważ z LUB logika.

Lokalizacja danych

Gdy wykonawca Spark jest współlokowany z serwerami regionu HBase, lokalizacja danych jest osiągana przez zidentyfikowanie lokalizacji serwera regionu i dokłada wszelkich starań, aby zlokalizować zadanie razem z serwerem regionu. Każdy executor wykonuje Scan/BulkGet na części danych znajdujących się na tym samym hoście.

Skanuj i pobieraj zbiorczo

Te dwa operatory są widoczne dla użytkowników przez określenie WHERE CLAUSE, np. WHERE kolumna> x i kolumna for scan i WHERE column =x zapominać. Operacje są wykonywane w executorach, a sterownik tylko je konstruuje. Wewnętrznie są one konwertowane na skanowanie i/lub pobieranie, a Iterator[Row] jest zwracany do silnika katalizatora w celu przetwarzania górnej warstwy.

[code language="scala"]def catalog = s"""{
         |"table":{"namespace":"default", "name":"table1"},
         |"rowkey":"key",
         |"columns":{
           |"col0":{"cf":"rowkey", "col":"key", "type":"string"},
           |"col1":{"cf":"cf1", "col":"col1", "type":"boolean"},
           |"col2":{"cf":"cf2", "col":"col2", "type":"double"},
           |"col3":{"cf":"cf3", "col":"col3", "type":"float"},
           |"col4":{"cf":"cf4", "col":"col4", "type":"int"},
           |"col5":{"cf":"cf5", "col":"col5", "type":"bigint"},
           |"col6":{"cf":"cf6", "col":"col6", "type":"smallint"},
           |"col7":{"cf":"cf7", "col":"col7", "type":"string"},
           |"col8":{"cf":"cf8", "col":"col8", "type":"tinyint"}
         |}
       |}""".stripMargin
[/code]