Sehr geehrte Leser, guten Tag!
In diesem Artikel beschreibt der leitende Berater der GeschĂ€ftssparte Big Data Solutions des Unternehmens âNeoflexâ ausfĂŒhrlich die Möglichkeiten, dynamische Strukturen mit Apache Spark zu erstellen.
Im Rahmen eines Datenanalyseprojekts taucht oft die Aufgabe auf, Frontendansichten auf der Basis von schwach strukturierten Daten zu erstellen.
In der Regel handelt es sich dabei um Protokolle oder Antworten verschiedener Systeme, die im JSON- oder XML-Format gespeichert werden. Die Daten werden in Hadoop exportiert, und anschlieĂend muss eine Frontendansicht erstellt werden. Den Zugriff auf die erstellte Ansicht können wir beispielsweise ĂŒber Impala organisieren.
In diesem Fall ist das Schema der Zielansicht im Voraus unbekannt. DarĂŒber hinaus kann das Schema auch nicht im Voraus erstellt werden, da es von den Daten abhĂ€ngt, und wir haben es mit exakt diesen schwach strukturierten Daten zu tun.
Zum Beispiel wird heute folgende Antwort protokolliert:
{source: "app1", error_code: ""}und morgen kommt von diesem System folgende Antwort:
{source: "app1", error_code: "error", description: "Netzwerkfehler"}In der Folge muss in die Frontendansicht ein weiteres Feld â description â hinzugefĂŒgt werden, und ob es kommt oder nicht, weiĂ niemand.
Die Aufgabe, eine Ansicht fĂŒr solche Daten zu erstellen, ist recht standardmĂ€Ăig, und Spark hat dafĂŒr eine Reihe von Werkzeugen. Es gibt UnterstĂŒtzung fĂŒr das Parsen der Ausgangsdaten sowohl in JSON als auch in XML, und fĂŒr ein im Voraus unbekanntes Schema ist schemaEvolution vorgesehen.
Auf den ersten Blick scheint die Lösung einfach zu sein. Man muss den Ordner mit JSON-Dateien nehmen und in ein DataFrame lesen. Spark erstellt das Schema und verwandelt die verschachtelten Daten in Strukturen. Danach mĂŒssen alle Daten im Parquet-Format gespeichert werden, das auch in Impala unterstĂŒtzt wird, indem die Ansicht im Hive Metastore registriert wird.
Es scheint also alles einfach zu sein.
Allerdings ist aus den kurzen Beispielen in der Dokumentation nicht klar, wie man in der Praxis mit einer Reihe von Problemen umgeht.
In der Dokumentation wird ein Ansatz beschrieben, der nicht zur Erstellung einer Ansicht dient, sondern zum Lesen von JSON oder XML in ein DataFrame.
Genauer gesagt, es wird einfach gezeigt, wie man JSON liest und parst:
df = spark.read.json(path...)Das reicht aus, um die Daten fĂŒr Spark verfĂŒgbar zu machen.
In der Praxis ist das Szenario viel komplizierter als nur JSON-Dateien aus einem Ordner zu lesen und ein DataFrame zu erstellen. Die Situation sieht so aus: Es gibt bereits eine bestimmte Ansicht, jeden Tag kommen neue Daten, und sie mĂŒssen in die Ansicht hinzugefĂŒgt werden, wobei nicht vergessen werden darf, dass das Schema abweichen kann.
Das ĂŒbliche Schema zum Erstellen einer Ansicht sieht so aus:
Schritt 1. Die Daten werden in Hadoop hochgeladen und tÀglich nachgeladen, wobei sie in eine neue Partition abgelegt werden. Es entsteht ein nach Tagen partitionierter Ordner mit den Quelldaten.
Schritt 2. WĂ€hrend des initialen Ladevorgangs wird dieser Ordner von Spark gelesen und geparst. Der resultierende DataFrame wird in ein fĂŒr die Analyse verfĂŒgbares Format, wie Parquet, gespeichert, das spĂ€ter in Impala importiert werden kann. So entsteht ein Ziel-Data Warehouse mit all den Daten, die bis zu diesem Zeitpunkt angesammelt wurden.
Schritt 3. Es wird ein Ladevorgang eingerichtet, der die Datenbank tÀglich aktualisiert.
Es stellt sich die Frage der inkrementellen Ladung, die Notwendigkeit der Partitionierung des Data Warehouses und die UnterstĂŒtzung des Gesamtschemas des Data Warehouses.
Nehmen wir ein Beispiel. Angenommen, der erste Schritt zum Aufbau des Lagers wurde umgesetzt und der Export von JSON-Dateien in einen Ordner konfiguriert.
Es ist kein Problem, daraus einen DataFrame zu erstellen, um ihn spÀter als Data Warehouse zu speichern. Dies ist der erste Schritt, den man leicht in der Spark-Dokumentation finden kann:
df = spark.read.option("mergeSchema", True).json(".../*")
df.printSchema()
root
|-- a: long (nullable = true)
|-- b: string (nullable = true)
|-- c: struct (nullable = true) |
|-- d: long (nullable = true)Es scheint alles gut zu sein.
Wir haben das JSON gelesen und geparst und speichern nun den DataFrame als Parquet, wobei wir ihn auf beliebige Weise in Hive registrieren:
df.write.format(âparquetâ).option('path','').saveAsTable('') Wir erhalten das Data Warehouse.
Am nĂ€chsten Tag sind neue Daten aus der Quelle hinzugekommen. Wir haben einen Ordner mit JSON und das Data Warehouse, das auf Grundlage dieses Ordners erstellt wurde. Nach dem Laden der nĂ€chsten Datenmenge fehlen im Data Warehouse die Daten fĂŒr einen Tag.
Eine logische Lösung wĂ€re es, das Data Warehouse nach Tagen zu partitionieren, was es ermöglicht, jeden folgenden Tag eine neue Partition hinzuzufĂŒgen. Auch dieser Mechanismus ist gut bekannt; Spark ermöglicht es, Partitionen separat zu speichern.
Zuerst fĂŒhren wir den initialen Ladevorgang durch und speichern die Daten wie oben beschrieben, wobei wir nur die Partitionierung hinzufĂŒgen. Dieser Vorgang wird als Initialisierung des Data Warehouses bezeichnet und erfolgt nur einmal:
df.write.partitionBy("date_load").mode("overwrite").parquet(dbpath + "\/" + db + "\/" + destTable)
Am nÀchsten Tag laden wir nur die neue Partition:
df.coalesce(1).write.mode("overwrite").parquet(dbpath + "\/" + db + "\/" + destTable +"\/date_load=" + date_load + "\/"))
Es bleibt nur noch, in Hive neu zu registrieren, um das Schema zu aktualisieren.
Hier treten jedoch Probleme auf.
Das erste Problem. FrĂŒher oder spĂ€ter kann das entstandene Parquet nicht gelesen werden. Das liegt daran, wie unterschiedlich Parquet und JSON mit leeren Feldern umgehen.
Betrachten wir eine typische Situation. Zum Beispiel kommt gestern JSON:
Tag 1: {"a": {"b": 1}},
und heute sieht dasselbe JSON so aus:
Tag 2: {"a": null}
Angenommen, wir haben zwei verschiedene Partitionen, in denen jeweils eine Zeile enthalten ist.
Wenn wir die Quelldaten komplett lesen, kann Spark den Typ bestimmen und versteht, dass âaâ ein Feld des Typs âStrukturâ ist, mit dem untergeordneten Feld âbâ vom Typ INT. Wenn jedoch jede Partition separat gespeichert wurde, ergibt sich ein Parquet mit inkompatiblen Schemas der Partitionen:
df1 (a: <struct>)
df2 (a: STRING NULLABLE)
Diese Situation ist gut bekannt, weshalb eine spezielle Option hinzugefĂŒgt wurde â beim Parsen der Quelldaten leere Felder zu entfernen:
df = spark.read.json("...", dropFieldIfAllNull=True)
In diesem Fall wird das Parquet aus Partitionen bestehen, die zusammen lesbar sind.
Obwohl diejenigen, die dies in der Praxis getan haben, hier bitter lĂ€cheln werden. Warum? Weil wahrscheinlich noch zwei Situationen entstehen werden. Oder drei. Oder vier. Die erste, die fast garantiert auftreten wird, ist, dass numerische Typen in verschiedenen JSON-Dateien unterschiedlich aussehen werden. Zum Beispiel, {intField: 1} und {intField: 1.1}. Wenn solche Felder in derselben Partition auftauchen, wird das Mergen des Schemas alles richtig lesen und zum genauesten Typ fĂŒhren. Wenn sie jedoch in unterschiedlichen Partitionen sind, wird in einer intField: int sein, und in der anderen intField: double.
Um mit dieser Situation umzugehen, gibt es das folgende Flag:
df = spark.read.json("...", dropFieldIfAllNull=True, primitivesAsString=True)
Jetzt haben wir einen Ordner, in dem sich die Partitionen befinden, die in einen einheitlichen DataFrame und ein gĂŒltiges Parquet der gesamten Ansicht gelesen werden können. Ja? Nein.
Man muss daran denken, dass wir die Tabelle in Hive registriert haben. Hive ist nicht case-sensitive gegenĂŒber den Feldnamen, wĂ€hrend Parquet das ist. Daher sind Partitionen mit Schemas: field1: int und Field1: int fĂŒr Hive identisch, nicht jedoch fĂŒr Spark. Man muss die Feldnamen in Kleinbuchstaben umwandeln.
Nach diesem Schritt scheint alles gut zu sein.
Allerdings ist nicht alles so einfach. Es tritt das zweite, ebenfalls gut bekannte Problem auf. Da jede neue Partition separat gespeichert wird, befinden sich im Ordner der Partition Dienstdateien von Spark, z. B. das Erfolgsflag _SUCCESS. Dies fĂŒhrt zu einem Fehler beim Versuch, Parquet zu erstellen. Um dies zu vermeiden, muss die Konfiguration so eingestellt werden, dass Spark das Schreiben von Dienstdateien in den Ordner untersagt:
hadoopConf = sc._jsc.hadoopConfiguration()
hadoopConf.set("parquet.enable.summary-metadata", "false")
hadoopConf.set("mapreduce.fileoutputcommitter.marksuccessfuljobs", "false")
Es scheint, dass jeden Tag eine neue Parquet-Partition in den Zielverzeichnis hinzugefĂŒgt wird, in der die geparsten Daten des Tages gespeichert sind. Wir haben vorausschauend dafĂŒr gesorgt, dass es keine Partitionen mit Typkonflikten gibt.
Doch nun stehen wir vor dem dritten Problem. Das allgemeine Schema ist unbekannt; darĂŒber hinaus hat die Hive-Tabelle eine falsche Schema, da jede neue Partition wahrscheinlich Verzerrungen ins Schema eingebracht hat.
Die Tabelle muss neu registriert werden. Das kann einfach gemacht werden: Wir lesen die Parquet-Daten des Verzeichnisses erneut, nehmen das Schema und erstellen darauf basierend das DDL, um den Ordner in Hive als externe Tabelle neu zu registrieren und das Schema des Zielverzeichnisses zu aktualisieren.
Vor uns steht das vierte Problem. Als wir die Tabelle zum ersten Mal registriert haben, haben wir uns auf Spark verlassen. Jetzt tun wir es selbst und mĂŒssen uns daran erinnern, dass Parquet-Felder mit Zeichen beginnen können, die in Hive nicht zulĂ€ssig sind. Zum Beispiel entfernt Spark die Zeilen, die es nicht in das Feld âcorrupt_recordâ parsen kann. Ein solches Feld kann in Hive nicht registriert werden, ohne es zu maskieren.
Wissen wir das, erhalten wir das Schema:
f_def = ""
for f in pf.dtypes:
if f[0] != "date_load":
f_def = f_def + "," + f[0].replace("_corrupt_record", "`_corrupt_record`") + " " + f[1].replace(":", "`:").replace("<", "<`").replace(",", ",`").replace("array<`", "array<")
table_define = "CREATE EXTERNAL TABLE jsonevolvtable (" + f_def[1:] + " ) "
table_define = table_define + "PARTITIONED BY (date_load string) STORED AS PARQUET LOCATION '\/user\/admin\/testJson\/testSchemaEvolution\/pq\/'"
hc.sql("drop table if exists jsonevolvtable")
hc.sql(table_define)
Code ("_corrupt_record", "`_corrupt_record`") + " " + f[1].replace(":", "`:").replace("<", "<`").replace(",", ",`").replace("array<`", "array<") macht sicheres DDL, das heiĂt anstelle von:
create table tname (_field1 string, 1field string)
Mit solchen Feldnamen wie â_field1, 1fieldâ wird sicheres DDL erstellt, bei dem die Feldnamen maskiert sind: create table `tname` (`_field1` string, `1field` string).
Es stellt sich die Frage: Wie erhĂ€lt man korrekt ein DataFrame mit vollstĂ€ndigem Schema (im Code pf)? Wie bekommt man dieses pf? Dies ist das fĂŒnfte Problem. Soll man das Schema aller Partitionen aus dem Ordner mit den Parquet-Dateien des Zielverzeichnisses erneut lesen? Diese Methode ist die sicherste, aber auch die schwerste.
Das Schema ist bereits in Hive vorhanden. Um ein neues Schema zu erhalten, muss das Schema der gesamten Tabelle mit dem der neuen Partition kombiniert werden. Das bedeutet, dass das Tabellenschema aus Hive genommen und mit dem Schema der neuen Partition vereint werden muss. Dies kann erreicht werden, indem die Testmetadaten aus Hive gelesen, in einem temporÀren Ordner gespeichert und dann mit Spark beide Partitionen gleichzeitig gelesen werden.
Im Grunde gibt es alles, was nötig ist: das ursprĂŒngliche Tabellenschema in Hive und die neue Partition. Die Daten liegen uns ebenfalls vor. Es bleibt nur, das neue Schema zu erhalten, in dem das Schema der Ansicht und die neuen Felder aus der erstellten Partition kombiniert werden:
from pyspark.sql import HiveContext
from pyspark.sql.functions import lit
hc = HiveContext(spark)
df = spark.read.json("...", dropFieldIfAllNull=True)
df.write.mode("overwrite").parquet(".../date_load=12-12-2019")
pe = hc.sql("select * from jsonevolvtable limit 1")
pe.write.mode("overwrite").parquet(".../fakePartiton/")
pf = spark.read.option("mergeSchema", True).parquet(".../date_load=12-12-2019/*", ".../fakePartiton/*")
Danach erstellen wir die DDL zur Registrierung der Tabelle, wie im vorherigen Fragment.
Wenn die gesamte Kette korrekt funktioniert, d.h. eine initiale Ladung erfolgt ist und die Tabelle in Hive korrekt erstellt wurde, erhalten wir das aktualisierte Tabellenschema.
Das letzte Problem besteht darin, dass eine Partition nicht einfach so zur Hive-Tabelle hinzugefĂŒgt werden kann, da sie sonst beschĂ€digt wird. Es ist notwendig, Hive dazu zu bringen, seine Partitionierungsstruktur zu reparieren:
from pyspark.sql import HiveContext
hc = HiveContext(spark)
hc.sql("MSCK REPAIR TABLE " + db + "." + destTable)
Die einfache Aufgabe, JSON zu lesen und darauf basierend eine Ansicht zu erstellen, entpuppt sich als Ăberwindung einer Reihe von impliziten Schwierigkeiten, fĂŒr deren Lösungen man separat suchen muss. Und obwohl diese Lösungen einfach sind, kostet ihre Suche viel Zeit.
Um die Ansicht zu erstellen, mussten wir:
- Partitionen zur Ansicht hinzufĂŒgen, wĂ€hrend wir uns von Systemdateien befreiten
- Mit leeren Feldern in den Quelldaten umgehen, die Spark typisiert hat
- Einfache Typen in Strings umwandeln
- Feldnamen in Kleinbuchstaben konvertieren
- Datenexport und Registrierung der Tabelle in Hive (DDL-Erstellung) trennen
- Nicht vergessen, Feldnamen zu escapen, die mit Hive möglicherweise inkompatibel sind
- Lernen, die Registrierung der Tabelle in Hive zu aktualisieren
Zusammenfassend lĂ€sst sich sagen, dass die Lösung zum Erstellen von Ansichten viele TĂŒcken birgt. Daher ist es besser, bei Schwierigkeiten auf einen erfahrenen Partner mit erfolgreicher Expertise zurĂŒckzugreifen.
Danke fĂŒr das Lesen dieses Artikels, wir hoffen, dass die Informationen nĂŒtzlich sind.
Quelle: habr.com
60GB SSD 8Gb DDR4
