Sehr geehrte Leser, guten Tag!
In diesem Artikel beschreibt der leitende Berater des Geschäftsbereichs Big Data Solutions bei «Neoflex» ausführlich die Möglichkeiten zur Erstellung von variabel strukturierten Data Warehouses mit Apache Spark.
Im Rahmen von Datenanalyseprojekten steht häufig die Aufgabe im Vordergrund, Data Warehouses aus schwach strukturierten Daten zu erstellen.
Dies sind in der Regel Protokolle oder Antworten verschiedener Systeme, die im JSON- oder XML-Format gespeichert werden. Die Daten werden in Hadoop exportiert, und aus ihnen muss ein Data Warehouse aufgebaut werden. Der Zugang zum erstellten Warehouse kann beispielsweise über Impala organisiert werden.
In diesem Fall ist das Schema des Ziel-Warehouse im Voraus unbekannt. Darüber hinaus kann das Schema auch nicht im Voraus erstellt werden, da es von den Daten abhängt und wir es mit diesen schwach strukturierten Daten zu tun haben.
Zum Beispiel wird heute folgende Antwort protokolliert:
{source: "app1", error_code: ""}Während am nächsten Tag von demselben System folgende Antwort kommt:
{source: "app1", error_code: "error", description: "Netzwerkfehler"}In der Folge sollte ein weiteres Feld — description — zum Data Warehouse hinzugefügt werden, doch ob es kommt oder nicht, weiß niemand.
Die Erstellung eines Dashboards mit solchen Daten ist eine gängige Aufgabe, und Spark bietet dafür eine Vielzahl von Werkzeugen. Es gibt Unterstützung für die Verarbeitung sowohl von JSON als auch von XML, und für unbekannte Schemata wird schemaEvolution unterstützt.
Auf den ersten Blick erscheint die Lösung einfach. Man muss einen Ordner mit JSON-Dateien nehmen und in einen DataFrame lesen. Spark wird ein Schema erstellen und verschachtelte Daten in Strukturen umwandeln. Danach müssen alle Daten in Parquet gespeichert werden, das unter anderem auch von Impala unterstützt wird, indem man das Dashboard im Hive Metastore registriert.
Es scheint alles einfach zu sein.
Allerdings wird aus den kurzen Beispielen in der Dokumentation nicht klar, wie man mit verschiedenen praktischen Problemen umgeht.
In der Dokumentation wird der Ansatz für das Lesen von JSON oder XML in einen DataFrame beschrieben, nicht für die Erstellung eines Dashboards.
Konkreter 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 jedoch weitaus komplexer, als nur JSON-Dateien aus einem Ordner zu lesen und einen DataFrame zu erstellen. Die Situation ist so: Es gibt bereits ein festgelegtes Dashboard, jeden Tag kommen neue Daten, die in das Dashboard integriert werden müssen, wobei man nicht vergessen darf, dass das Schema unterschiedlich sein kann.
Das gängige Schema für den Aufbau einer Data Warehouse-Ansicht ist wie folgt:
Schritt 1. Daten werden in Hadoop geladen, gefolgt von täglichen Aktualisierungen, und in eine neue Partition eingeordnet. Es entsteht ein nach Tagen partitionierter Ordner mit den Rohdaten.
Schritt 2. Im Rahmen der initialen Datenladung wird dieser Ordner gelesen und mit Spark verarbeitet. Der resultierende DataFrame wird in einem analysierbaren Format gespeichert, beispielsweise in Parquet, das später in Impala importiert werden kann. So wird eine Zielansicht mit allen zu diesem Zeitpunkt angesammelten Daten erstellt.
Schritt 3. Es wird ein Prozess eingerichtet, der die Ansicht täglich aktualisiert.
Es stellt sich die Frage nach der inkrementellen Ladeprozedur, der Notwendigkeit der Partitionierung der Ansicht und der Unterstützung des allgemeinen Schemas der Ansicht.
Nehmen wir als Beispiel an, dass der erste Schritt zum Bau des Speichers implementiert ist und die Exportierung von JSON-Dateien in einen Ordner eingerichtet wurde.
Es ist kein Problem, aus ihnen einen DataFrame zu erstellen, um ihn später als Ansicht zu speichern. Das 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)Scheint alles in Ordnung zu sein.
Wir haben das JSON gelesen und geparsed, danach speichern wir das DataFrame als Parquet und registrieren es in Hive auf beliebige Weise:
df.write.format("parquet").option('path','').saveAsTable('') Wir erhalten die Sicht.
Am nächsten Tag sind neue Daten aus der Quelle hinzugekommen. Wir haben einen Ordner mit JSON und eine Sicht, die auf diesem Ordner basiert. Nach dem Laden der nächsten Datenmenge aus der Quelle fehlen in der Sicht die Daten für einen Tag.
Eine logische Lösung wäre, die Sicht nach Tagen zu partitionieren, was es ermöglicht, jeden folgenden Tag eine neue Partition hinzuzufügen. Dieses Verfahren ist ebenfalls gut bekannt, Spark erlaubt es, Partitionen separat zu schreiben.
Zuerst führen wir einen Initial-Ladevorgang durch, indem wir die Daten wie oben beschrieben speichern und nur die Partitionierung hinzufügen. Diese Handlung wird als Initialisierung der Sicht 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 wird das erzeugte Parquet nicht mehr lesbar sein. Das hängt damit zusammen, wie unterschiedlich Parquet und JSON Leerwerte behandeln.
Betrachten wir eine typische Situation. Zum Beispiel kam gestern folgendes JSON an:
Tag 1: {"a": {"b": 1}},
und heute sieht dieses JSON so aus:
Tag 2: {"a": null}
Angenommen, wir haben zwei verschiedene Partitionen, in denen jeweils eine Zeile vorhanden ist.
Wenn wir die ursprünglichen Daten als Ganzes lesen, kann Spark den Typ bestimmen und versteht, dass „a“ ein Feld des Typs „Struktur“ ist, mit einem untergeordneten Feld „b“ vom Typ INT. Wenn jedoch jede Partition separat gespeichert wurde, ergibt sich Parquet mit inkompatiblen Partition-Schemas:
df1 (a: <struct>)
df2 (a: STRING NULLABLE)
Diese Situation ist gut bekannt, daher wurde absichtlich die Option hinzugefügt – beim Parsen der Ausgangsdaten 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 gemacht haben, bitter lächeln werden. Warum? Weil wahrscheinlich noch zwei Situationen auftreten werden. Oder drei. Oder vier. Die erste, die fast sicher auftritt, ist, dass numerische Typen in verschiedenen JSON-Dateien unterschiedlich aussehen werden. Zum Beispiel {intField: 1} und {intField: 1.1}. Wenn solche Felder in einem Partition vorkommen, wird das Mergen der Schemata alles richtig lesen, wodurch der genaueste Typ erreicht wird. Sollte es jedoch in verschiedenen Partitionen sein, 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 Vitrine gelesen werden können. Oder? Nein.
Man muss daran denken, dass wir die Tabelle in Hive registriert haben. Hive ist nicht empfindlich gegenüber Groß- und Kleinschreibung bei Feldnamen, während Parquet empfindlich ist. Daher sind Partitionen mit den Schemata: field1: int und Field1: int für Hive gleich, aber nicht für Spark. Man muss nicht vergessen, die Feldnamen in Kleinbuchstaben zu bringen.
Es scheint, dass nach diesem alles gut ist.
Es ist jedoch nicht so einfach. Es tritt ein zweites, ebenfalls gut bekanntes Problem auf. Da jede neue Partition separat gespeichert wird, befinden sich in dem Partitionen-Ordner Verwaltungsdateien von Spark, beispielsweise die Erfolgsflagge _SUCCESS. Dies führt zu einem Fehler beim Versuch, Parquet zu verwenden. Um dies zu vermeiden, muss die Konfiguration so eingestellt werden, dass Spark das Schreiben von Verwaltungsdateien in den Ordner verhindert:
hadoopConf = sc._jsc.hadoopConfiguration()
hadoopConf.set("parquet.enable.summary-metadata", "false")
hadoopConf.set("mapreduce.fileoutputcommitter.marksuccessfuljobs", "false")
Es scheint, als würde nun jeden Tag eine neue Parquet-Partition in den Zielordner hinzugefügt, in der die Daten des Tages gespeichert sind. Wir haben im Voraus dafür gesorgt, dass keine Partitionen mit Datentypkonflikten auftreten.
Doch nun steht uns das dritte Problem bevor. Jetzt ist das Gesamtschema unbekannt, zudem hat die Hive-Tabelle ein falsches Schema, da jede neue Partition wahrscheinlich eine Verzerrung im Schema verursacht hat.
Es ist notwendig, die Tabelle neu zu registrieren. Das kann ganz einfach gemacht werden: Lies erneut die Parquet-Vitrine, entnehme das Schema und erstelle auf dieser Basis ein DDL, mit dem der Ordner in Hive erneut als externe Tabelle registriert werden kann, wobei das Schema der Zielvitrine aktualisiert wird.
Vor uns steht das vierte Problem. Als wir die Tabelle das erste Mal registriert haben, haben wir uns auf Spark stützt. Jetzt machen wir das selbst und müssen uns daran erinnern, dass Parquet-Felder mit Zeichen beginnen können, die für Hive ungültig sind. Zum Beispiel entfernt Spark Zeilen, die es nicht in das Feld „corrupt_record“ parsen kann. Ein solches Feld kann nicht in Hive registriert werden, ohne es zu escapen.
Mit diesem Wissen 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 ein sicheres DDL, also anstelle von:
create table tname (_field1 string, 1field string)
Mit solchen Feldnamen wie „_field1, 1field“ wird ein sicheres DDL erstellt, bei dem die Feldnamen escaped werden: create table `tname` (`_field1` string, `1field` string).
Die Frage ist: Wie erhält man korrekt ein DataFrame mit dem vollständigen Schema (im Code pf)? Wie bekommt man dieses pf? Das ist das fünfte Problem. Sollten wir das Schema aller Partitionen aus dem Ordner mit den Parquet-Dateien des Ziel-Data-Warehouse neu lesen? Diese Methode ist die sicherste, aber auch die aufwändigste.
Das Schema ist bereits in Hive vorhanden. Um ein neues Schema zu erhalten, kann das gesamte Schema der Tabelle mit der neuen Partition kombiniert werden. Das bedeutet, das Schema der Tabelle aus Hive abzurufen und es mit dem Schema der neuen Partition zu vereinen. Dies kann erreicht werden, indem die Testmetadaten aus Hive gelesen und vorübergehend gespeichert werden, um dann mit Spark beide Partitionen gleichzeitig zu lesen.
Im Wesentlichen haben wir alles, was wir brauchen: das ursprüngliche Schema der Tabelle in Hive und die neue Partition. Die Daten sind ebenfalls vorhanden. Es bleibt nur, das neue Schema zu erhalten, in dem das Schema des Data Warehouses mit den neuen Feldern aus der erstellten Partition kombiniert wird.
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/*")
Anschließend erstellen wir das DDL zur Registrierungs der Tabelle, wie im vorherigen Fragment.
Wenn die gesamte Kette korrekt funktioniert, d.h. es gab einen initialen Ladeprozess und die Tabelle wurde korrekt in Hive erstellt, erhalten wir das aktualisierte Tabellenschema.
Und das letzte Problem besteht darin, dass man nicht einfach eine Partition zur Hive-Tabelle hinzufügen kann, da sie sonst beschädigt wird. Hive muss gezwungen werden, die Struktur der Partitionen zu reparieren:
from pyspark.sql import HiveContext
hc = HiveContext(spark)
hc.sql("MSCK REPAIR TABLE " + db + "." + destTable)
Die einfache Aufgabe, JSON zu lesen und auf dieser Basis ein Datenmodell zu erstellen, erweist sich als die Überwindung einer Reihe von impliziten Schwierigkeiten, für deren Lösungen man separat suchen muss. Auch wenn diese Lösungen einfach sind, kostet ihre Auffindung viel Zeit.
Um das Datenmodell zu erstellen, musste ich:
- Partitionen zum Datenmodell hinzufügen, indem ich die temporären Dateien entfernt habe
- Leerfelder in den Quelldaten klären, die Spark typisiert hat
- Einfache Datentypen in Strings konvertieren
- Feldnamen in Kleinschreibung umwandeln
- Den Datenexport und die Registrierung der Tabelle in Hive (DDL-Erstellung) trennen
- Nicht vergessen, die Feldnamen zu maskieren, die möglicherweise inkompatibel mit Hive sind
- Lernen, die Registrierung der Tabelle in Hive zu aktualisieren
Zusammenfassend lässt sich sagen, dass die Lösung zur Erstellung von Datenvitrinen viele Herausforderungen birgt. Daher ist es ratsam, bei Schwierigkeiten in der Umsetzung einen erfahrenen Partner mit erfolgreicher Expertise zu Rate zu ziehen.
Vielen Dank für das Lesen dieses Artikels. Wir hoffen, dass die Informationen nützlich sind.
Quelle: habr.com
Zuverlässiges Webhosting mit DDoS-Schutz, VPS- und VDS-Server kaufen 🔥 Zuverlässiges Webhosting mit DDoS-Schutz, VPS- und VDS-Server kaufen | ProHoster
