Spark-on-HBase:op DataFrame gebaseerde HBase-connector

Deze blogpost is gepubliceerd op Hortonworks.com vóór de fusie met Cloudera. Sommige links, bronnen of verwijzingen zijn mogelijk niet langer nauwkeurig.

Met trots kondigen we de technische preview aan van Spark-HBase Connector, ontwikkeld door Hortonworks in samenwerking met Bloomberg.

De Spark-HBase-connector maakt gebruik van Data Source API (SPARK-3247) die is geïntroduceerd in Spark-1.2.0. Het overbrugt de kloof tussen de eenvoudige HBase Key Value-opslag en complexe relationele SQL-query's en stelt gebruikers in staat om complexe gegevensanalyses uit te voeren bovenop HBase met behulp van Spark. Een HBase DataFrame is een standaard Spark DataFrame en kan communiceren met andere gegevensbronnen zoals Hive, ORC, Parquet, JSON, enz.

Achtergrond

Er zijn verschillende open source Spark HBase-connectoren beschikbaar als Spark-pakketten, als onafhankelijke projecten of in HBase-trunk.

Spark is verhuisd naar de Dataset/DataFrame-API's, die ingebouwde optimalisatie van queryplannen bieden. Nu geven eindgebruikers er de voorkeur aan een op DataFrames/Datasets gebaseerde interface te gebruiken.

De HBase-connector in de HBase-trunk heeft een rijke ondersteuning op RDD-niveau, b.v. BulkPut, enz., maar de DataFrame-ondersteuning is niet zo uitgebreid. De HBase-trunkconnector is afhankelijk van de standaard HadoopRDD met HBase ingebouwd TableInputFormat heeft enkele prestatiebeperkingen. Bovendien kan BulkGet uitgevoerd in de driver een single point of failure zijn.

Er zijn enkele andere alternatieve implementaties. Neem Spark-SQL-on-HBase als voorbeeld. Het past zeer geavanceerde aangepaste optimalisatietechnieken toe door zijn eigen query-optimalisatieplan in te bedden in de standaard Spark Catalyst-engine, verzendt de RDD naar HBase en voert gecompliceerde taken uit, zoals gedeeltelijke aggregatie, in de HBase-coprocessor. Deze aanpak is in staat om hoge prestaties te bereiken, maar het is moeilijk te handhaven vanwege de complexiteit en de snelle evolutie van Spark. Ook het toestaan ​​dat willekeurige code in een coprocessor wordt uitgevoerd, kan beveiligingsrisico's met zich meebrengen.

De Spark-on-HBase Connector (SHC) is ontwikkeld om deze potentiële knelpunten en zwakke punten te verhelpen. Het implementeert de standaard Spark Datasource-API en maakt gebruik van de Spark Catalyst-engine voor query-optimalisatie. Tegelijkertijd wordt de RDD helemaal opnieuw opgebouwd in plaats van TableInputFormat . te gebruiken om hoge prestaties te bereiken. Met deze op maat gemaakte RDD kunnen alle kritische technieken worden toegepast en volledig geïmplementeerd, zoals partitie pruning, column pruning, predicaat pushdown en data locality. Het ontwerp maakt het onderhoud zeer eenvoudig, terwijl een goede balans tussen prestatie en eenvoud wordt bereikt.

Architectuur

We gaan ervan uit dat Spark en HBase in hetzelfde cluster worden geïmplementeerd en dat Spark-uitvoerders zich op dezelfde locatie bevinden als regioservers, zoals geïllustreerd in de onderstaande afbeelding.

Afbeelding 1. Spark-on-HBase-connectorarchitectuur

Op een hoog niveau behandelt de connector zowel Scannen als ophalen op een vergelijkbare manier, en beide acties worden uitgevoerd in de uitvoerders. De driver verwerkt de query, aggregeert scans/gets op basis van de metadata van de regio en genereert taken per regio. De taken worden verzonden naar de voorkeursuitvoerders die zich samen met de regioserver bevinden, en worden parallel uitgevoerd in de uitvoerders om een ​​betere gegevenslokaliteit en gelijktijdigheid te bereiken. Als een regio niet over de vereiste gegevens beschikt, krijgt die regioserver geen taak toegewezen. Een taak kan bestaan ​​uit meerdere Scans en BulkGets, en de gegevensverzoeken door een taak worden opgehaald van slechts één regioserver, en deze regioserver zal ook de locatievoorkeur voor de taak zijn. Merk op dat de bestuurder niet betrokken is bij de echte taakuitvoering, behalve bij het plannen van taken. Dit voorkomt dat de bestuurder de bottleneck wordt.

Tabelcatalogus

Om de HBase-tabel als relationele tabel in Spark te brengen, definiëren we een toewijzing tussen HBase- en Spark-tabellen, genaamd Table Catalog. Er zijn twee kritieke delen van deze catalogus. Een daarvan is de rowkey-definitie en de andere is de toewijzing tussen tabelkolom in Spark en de kolomfamilie en kolomkwalificatie in HBase. Raadpleeg het gedeelte Gebruik voor details.

Native Avro-ondersteuning

De connector ondersteunt native de Avro-indeling, omdat het heel gebruikelijk is om gestructureerde gegevens als bytearray in HBase te bewaren. De gebruiker kan de Avro-record rechtstreeks in HBase bewaren. Intern wordt het Avro-schema automatisch geconverteerd naar een native Spark Catalyst-gegevenstype. Houd er rekening mee dat beide sleutel/waarde-onderdelen in een HBase-tabel kunnen worden gedefinieerd in Avro-indeling. Raadpleeg de voorbeelden/testcases in de repo voor het exacte gebruik.

Predikaat Pushdown

De connector haalt alleen vereiste kolommen op van de regioserver om de netwerkoverhead te verminderen en redundante verwerking in de Spark Catalyst-engine te voorkomen. Bestaande standaard HBase-filters worden gebruikt om predikaat-push-down uit te voeren zonder gebruik te maken van de coprocessor-capaciteit. Omdat HBase het gegevenstype niet kent, behalve byte-array, en de volgorde-inconsistentie tussen Java-primitieve typen en byte-array, moeten we de filtervoorwaarde vooraf verwerken voordat het filter in de scanbewerking wordt ingesteld om gegevensverlies te voorkomen. Binnen de regioserver worden records die niet overeenkomen met de queryvoorwaarde eruit gefilterd.

Partitie snoeien

Door de rijsleutel uit de predikaten te extraheren, splitsen we de Scan/BulkGet op in meerdere niet-overlappende bereiken, alleen de regioservers met de gevraagde gegevens zullen Scan/BulkGet uitvoeren. Momenteel wordt het snoeien van partities uitgevoerd op de eerste dimensie van de rijsleutels. Als een rijsleutel bijvoorbeeld "key1:key2:key3" is, wordt het opschonen van partities alleen gebaseerd op "key1". Merk op dat de WHERE-voorwaarden zorgvuldig moeten worden gedefinieerd. Anders wordt het snoeien van partities mogelijk niet van kracht. Bijvoorbeeld, WHERE rowkey1> "abc" OR column ="xyz" (waar rowkey1 de eerste dimensie van de rowkey is en column een normale hbase-kolom) resulteert in een volledige scan, omdat we alle bereiken moeten bestrijken omdat van de OF logica.

Gegevenslocatie

Wanneer een Spark-uitvoerder zich op dezelfde locatie bevindt met HBase-regioservers, wordt de gegevenslocatie bereikt door de locatie van de regioserver te identificeren en wordt er alles aan gedaan om de taak samen met de regioserver te plaatsen. Elke uitvoerder voert Scan/BulkGet uit van de kant van de gegevens die zich op dezelfde host bevinden.

Scannen en bulksgewijs ophalen

Deze twee operators zijn zichtbaar voor gebruikers door WHERE CLAUSE op te geven, bijvoorbeeld WHERE column> x en column voor scan en WAAR kolom =x vergeten. De bewerkingen worden uitgevoerd in de uitvoerders en de bestuurder construeert alleen deze bewerkingen. Intern worden ze geconverteerd naar scannen en/of get, en Iterator[Row] wordt teruggestuurd naar de katalysatormotor voor verwerking van de bovenste laag.

[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]