Introduktion
Python bruges flittigt blandt dataingeniører og dataforskere til at løse alle mulige problemer fra ETL/ELT-pipelines til opbygning af maskinlæringsmodeller. Apache HBase er et effektivt datalagringssystem til mange arbejdsgange, men at få adgang til disse data specifikt gennem Python kan være en kamp. For dataprofessionelle, der ønsker at gøre brug af data, der er lagret i HBase, kan det seneste upstream-projekt "hbase-connectors" bruges med PySpark til grundlæggende operationer.
I denne blogserie vil vi forklare, hvordan man konfigurerer PySpark og HBase sammen til grundlæggende Spark-brug såvel som til job, der vedligeholdes i CDSW. For dem, der ikke er bekendt med CDSW, er det en sikker, selvbetjent virksomhedsdatavidenskabsplatform, hvor dataforskere kan administrere deres egne analysepipelines og dermed accelerere maskinlæringsprojekter fra udforskning til produktion. Besøg Cloudera Data Science Workbench-produktsiden for mere information om CDSW.
I dette indlæg vil flere operationer blive forklaret og demonstreret sammen med eksempel output. Til kontekst køres alle eksempelhandlinger i dette specifikke blogindlæg med en CDSW-implementering.
Forudsætninger:
- Har en CDP-klynge med HBase og Spark
- Hvis du vil følge eksempler via CDSW, skal du have det installeret – Installation af Cloudera Data Science Workbench
- Python 3 er installeret på hver node på samme sti
Konfiguration:
Først skal HBase og Spark konfigureres sammen, for at Spark SQL-forespørgsler kan fungere korrekt. For at gøre det er der to dele til det:Først skal du konfigurere HBase-regionsserverne gennem Cloudera Manager; og for det andet, sørg for, at Spark-løbetiden har HBase-bindinger. En bemærkning at huske på er, at Cloudera Manager allerede opsætter nogle konfigurations- og miljøvariabler for automatisk at pege Spark mod HBase for dig. Ikke desto mindre er det første trin i konfigurationen af Spark SQL-forespørgsler almindeligt gennem alle typer implementering på CDP-klynger, men det andet er lidt anderledes afhængigt af typen af implementering.
Konfiguration af HBase-regionsservere
- Gå til Cloudera Manager, og vælg HBase-tjenesten.
- Søg efter "regionserver-miljø"
- Tilføj en ny miljøvariabel ved hjælp af RegionServer Environment Advanced Configuration Snippet (sikkerhedsventil):
- Nøgle:HBASE_CLASSPATH
- Værdi:/opt/cloudera/parcels/CDH/lib/hbase_connectors/lib/hbase-spark.jar:/opt/cloudera/parcels/CDH/lib/hbase_connectors/lib /hbase-spark-protocol-shaded.jar:/opt/cloudera/parcels/CDH/jars/scala-library-2.11.12.jar
Sørg for, at du bruger de korrekte versionsnumre.
- Genstart regionsservere.
Når du har fulgt ovenstående trin, skal du følge trinene nedenfor afhængigt af, om du ønsker en CDSW- eller ikke-CDSW-implementering.
Tilføjelse af HBase-bindinger til Spark Runtime i ikke-CDSW-implementeringer
For at implementere skallen eller bruge spark-submit korrekt skal du bruge følgende kommandoer for at sikre, at gnisten har de rigtige HBase-bindinger.
pyspark –jars /opt/cloudera/parcels/CDH/lib/hbase_connectors/lib/hbase-spark.jar,/opt/cloudera/parcels/CDH/lib/hbase_connectors/lib/hbase-spark-protocol-shaded. krukke
spark-submit –jars /opt/cloudera/parcels/CDH/lib/hbase_connectors/lib/hbase-spark.jar,/opt/cloudera/parcels/CDH/lib/hbase_connectors/lib/hbase-spark-protocol- shaded.jar
Tilføjelse af HBase-bindinger til Spark Runtime i CDSW-installationer
For at konfigurere CDSW med HBase og PySpark er der et par trin, du skal tage.
1) Sørg for, at Python 3 er installeret på hver klynge node, og noter stien til den
2) Lav et nyt projekt i CDSW og brug en PySpark-skabelon
3) Åbn projektet, gå til Indstillinger -> Motor -> Miljøvariabler.
4) Indstil PYSPARK3_DRIVER_PYTHON og PYSPARK3_PYTHON til stien, hvor Python er installeret på dine klynge noder (sti noteret i trin 1).
Nedenfor er et eksempel på, hvordan det skal se ud.
5) I dit projekt skal du gå til Filer -> spark-defaults.conf og åbne det i Workbench
6) Kopiér og indsæt linjen nedenfor i den fil, og sørg for, at den er gemt, før du starter en ny session.
spark.jars=/opt/cloudera/parcels/CDH/lib/hbase_connectors/lib/hbase-spark.jar,/opt/cloudera/parcels/CDH/lib/hbase_connectors/lib/hbase-spark-protocol-shaded.jar
På dette tidspunkt er CDSW nu konfigureret til at køre PySpark-job på HBase! Resten af dette blogindlæg henviser til nogle eksempler på operationer på en CDSW-installation.
Eksempel på handlinger
Put Operations
Der er to måder at indsætte og opdatere rækker på i HBase. Den første og mest anbefalede metode er at bygge et katalog, som er et skema, der vil kortlægge kolonnerne i en HBase-tabel til en PySpark-dataramme, mens tabellens navn og navneområde specificeres. Opbygning af dette brugerdefinerede JSON-format er den mest foretrukne metode, da det også kan bruges med andre operationer. For mere information om kataloger henvises til denne dokumentation http://hbase.apache.org/book.html#_define_catalog. Den anden metode bruger en specifik tilknytningsparameter kaldet "hbase.columns.mapping", som blot tager en streng af nøgleværdi-par.
- Brug af kataloger
from pyspark.sql import Row
from pyspark.sql import SparkSession
spark = SparkSession\
.builder\
.appName("SampleApplication")\
.getOrCreate()
tableCatalog = ''.join("""{
"table":{"namespace":"default", "name":"tblEmployee", "tableCoder":"PrimitiveType"},
"rowkey":"key",
"columns":{
"key":{"cf":"rowkey", "col":"key", "type":"int"},
"empId":{"cf":"personal","col":"empId","type":"string"},
"empName":{"cf":"personal", "col":"empName", "type":"string"},
"empState":{"cf":"personal", "col":"empWeight", "type":"string"}
}
}""".split())
employee = [(10, 'jonD', 'Jon Daniels', 'CA'), (6, 'billR', 'Bill Robert', 'FL')]
employeeRDD = spark.sparkContext.parallelize(employee)
employeeMap = employeeRDD.map(lambda x: Row(key=int(x[0]), empId=x[1], empName=x[2], empState=x[3]))
employeeDF = spark.createDataFrame(employeeMap)
employeeDF.write.format("org.apache.hadoop.hbase.spark") \
.options(catalog=tableCatalog, newTable=5) \
.option("hbase.spark.use.hbasecontext", False) \
.save()
# newTable refers to the NumberOfRegions which has to be > 3 Bekræft, at en ny tabel kaldet "tblEmployee" er oprettet i HBase ved blot at åbne HBase-skallen og udføre følgende kommando:
scan 'tblEmployee', {'LIMIT' => 2}
Brug af kataloger kan også give dig mulighed for nemt at indlæse HBase-tabeller. Dette vil blive diskuteret i en fremtidig udgave.
- Brug af hbase.columns.mapping
Mens du skriver PySpark Dataframe, kan en mulighed kaldet "hbase.columns.mapping" tilføjes for at inkludere en streng, der kortlægger kolonnerne korrekt. Denne mulighed giver dig kun mulighed for at indsætte rækker i eksisterende tabeller.
I HBase-skallen, lad os først oprette en tabel og oprette 'tblEmployee2', 'personal'
Lad os nu indsætte 2 rækker i PySpark ved hjælp af "hbase.columns.mapping"
from pyspark.sql import Row
from pyspark.sql import SparkSession
spark = SparkSession\
.builder\
.appName("SampleApplication")\
.getOrCreate()
employee = [(10, 'jonD', 'Jon Daniels', 170.7), (6, 'billR', 'Bill Robert', 200.1)]
employeeRDD = spark.sparkContext.parallelize(employee)
employeeMap = employeeRDD.map(lambda x: Row(key=int(x[0]), empId=x[1], empName=x[2], empWeight=float(x[3])))
employeeDF = spark.createDataFrame(employeeMap)
employeeDF.write.format("org.apache.hadoop.hbase.spark") \
.option("hbase.columns.mapping", "key INTEGER :key, empId STRING personal:empId, empName STRING personal:empName, empWeight FLOAT personal:empWeight") \
.option("hbase.table", "tblEmployee2") \
.option("hbase.spark.use.hbasecontext", False) \
.save() Igen skal du bare kontrollere, at en ny tabel kaldet "tblEmployee2" har disse nye rækker.
scan 'tblEmployee2', {'LIMIT' => 2}
Det fuldender vores eksempler på, hvordan man indsætter rækker gennem PySpark i HBase-tabeller. I den næste del vil jeg diskutere Get and Scan Operations, PySpark SQL og noget fejlfinding. Indtil da bør du anskaffe dig en CDP-klynge og arbejde dig igennem disse eksempler.