Methoden und Werkzeuge für die Datenaufbereitung im Big Data Bereich

All diese Methoden benötigen als Grundlage allerdings eine größere Menge an qualitativ hochwertigen und validen Daten. In diesem Kontext kommt Data Engineering zum Einsatz, um die Rohdaten zu sammeln, zu bereinigen und zu einer integrierten Datenbasis zusammenzuführen. Während in einem früheren Artikel die generelle Rolle und Ziele von Data Engineering beleuchtet wurden, soll hier der Fokus auf Methoden und bewährte Werkzeuge gelegt werden sowie ein beispielhafter Einblick in die algorithmische Umsetzung von Data Engineering Aufgaben gegeben werden.

Datenstrom- und Batch-Verarbeitung

Wenn beispielsweise industrielle Sensordaten über die Zeit gesammelt werden, fallen pro Zeiteinheit (zB. alle paar Sekunden) keine großen Datenmengen an, über Monate und Jahre gehen die gespeicherten Daten aber oft in den Terabyte-Bereich. Wenn Daten in dieser Größenordnung verarbeitet werden sollen, kann dies im Wesentlichen durch zwei verschiedene Paradigmen erfolgen, hier beschrieben für die Konvertierung des Datentypen einer Tabellenspalte:

• Batch-Verarbeitung:
  Hierbei werden alle Zeilen in einer Tabelle datenparallel auf einmal verarbeitet, um eine Spalte zu konvertieren.
• Datenstromverarbeitung (Data Streaming):
  Hierbei werden die Zeilen der Tabelle sequentiell eingelesen und die Spaltenkonvertierung pro Zeile durchgeführt.

Der wesentliche Unterschied zwischen den beiden Datenverarbeitungsansätzen ist, dass beim Data Streaming die notwendigen Datentransformationen – wie beispielsweise die Konvertierung von Datenfeldern auf andere Datentypen – direkt auf dem aktuell gelieferten Datensatz durchgeführt werden, während bei der Batch-Verarbeitung die Daten zuerst gesammelt werden, und in der Folge die Datentransformationen auf der Gesamtheit der Daten durchgeführt werden.

Welcher Ansatz gewählt wird, ist von den Anforderungen an die Datentransformation abhängig:

• Wenn die Transformation lokal auf den aktuell abgefragten oder empfangenen Daten durchgeführt werden kann, ist oft der Einsatz des Data Streaming Ansatzes vorzuziehen, da er meist eine einfachere und lokalere Operation darstellt, die aufgrund der geringeren Eingabedaten auch schneller abgearbeitet werden kann. Ein typischer Einsatzbereich von Data Streaming ist die direkte Konvertierung von über die Zeit verteilt eintreffenden Sensordaten, da diese dann einzeln konvertiert und gespeichert werden können.
• Wenn allerdings die Datentransformation Eingabedaten aus dem gesamten bereits gespeicherten Daten benötigt oder bereits alle Daten vorliegen, eignet sich ein Batch-Ansatz besser. Hierbei ist auch oft eine datenparallele Verarbeitung einfacher umzusetzen, da diese von Frameworks wie Apache Hadoop (durch den Map-Reduce Ansatz) oder Apache Spark direkt unterstützt wird.

Generell sollten die erhaltenen Daten einmal im Rohformat gespeichert werden, um keine Daten zu verlieren, die als Grundlage für zukunftige Analysen noch benötigt werden könnten. Die Weiterverarbeitung so gespeicherter Daten kann dann durch Batch-Verarbeitung oder Data Streaming erfolgen. Im zweiteren Fall wird aus den gespeicherten Daten durch kontinuierliches Lesen wieder ein Datenstrom erzeugt. Umgekehrt kann ein Datenstrom kontinuierlich gespeichert werden und somit als Ausgangsbasis für Batch-Verarbeitung dienen.

Datenstromverarbeitung: Apache NiFi

NiFi stellt ein Werkzeug zur Datenstromverarbeitung dar, welches es ermöglicht, in einer grafischen webbasierten Benutzerschnittstelle Datentransformationen zu einer durchgängigen Datenpipeline zu verbinden, durch die die Quelldaten fließen und schrittweise transformiert werden. Die Stärken von Apache NiFi liegen in der breiten Palette an bereits verfügbaren Modulen die beispielsweise das Einlesen und Speichern von zahlreichen Datenformaten ermöglichen. Durch den Open-Source Charakter von NiFi und die objektorientierte Struktur seiner Module ist es leicht möglich, selbst Module zu entwickeln und diese in Datenpipelines zu integrieren. Weiters werden durch NiFi auch Themen wie die automatisierte Behandlung von verschiedenen Verarbeitungsgeschwindigkeiten der Module geregelt.

Batch-Verarbeitung: Apache Hadoop

Hadoop ist ein Software-Framework, das auf dem Grundprinzip der datenparallelen Verarbeitung in einer Cluster-Umgebung beruht. Innerhalb der verteilten Verarbeitung übernimmt jeder Cluster-Rechner die Verarbeitung der dort lokal vorliegenden Daten, womit vor allem Kommunikationsaufwand während der Berechnungen eingespart werden kann. Hadoop unterscheidet hierbei zwischen Master- und Slave-Diensten im Cluster, wobei die Slave-Dienste die Verarbeitung der lokal vorliegenden Daten übernehmen, während den Master-Diensten die Koordination des Clusters obliegt. Teile der in Hadoop implementierten Algorithmen wurden von Google entwickelt und die Konzepte in Forschungspapers veröffentlicht, wie beispielsweise das Google File System, Map-Reduce und Google Bigtable. Bei Google werden diese Lösungen für den Betrieb der weltweiten Suchinfrastruktur eingesetzt, während das Hadoop-Projekt eine Open-Source Implementierung dieser Konzepte darstellt.

Im Kern besteht ein Hadoop-System aus einem üblicherweise linux-basierten Cluster, auf dem das Hadoop File System (HDFS) und YARN als Implementierung des Map-Reduce Algorithmus laufen. Ein Hadoop-Cluster mit den Diensten HDFS und Yarn stellt eine solide technologische Basis für verschiedenste Big-Data Dienste wie BigTable-Datenbanken wie HBase – siehe weiter unten – oder Graph-Datenbanken wie beispielsweise JanusGraph dar.

Hadoop File System (HDFS)

HDFS ist eine Open-Source Implementierung des Google Filesystems. Es besteht genauso wie andere Hadoop-Teilsysteme aus Master- und Slave-Komponenten, im Fall von HDFS Namenodes (Master) und Datanodes (Slaves). Während ein Namenode speichert, wo am Cluster die Daten für einzelnen Dateien hinterlegt sind, übernehmen die Datanodes die Speicherung der Datenblöcke. Grundsätzlich ist HDFS für große Dateien optimiert, die Blockgröße für die Speicherung beträgt üblicherweise 128 Megabyte. Einerseits kann eine Datei aus vielen einzelnen Blöcken bestehen, andererseits werden die Datenblöcke aus Redundanz- und Performancegründen über mehrere Clusterknoten repliziert. Die Zugriffssemantik von HDFS unterscheidet sich von der üblichen Posix-Semantik, da an HDFS-Dateien nur Daten angehängt werden können, diese aber nicht editiert werden können. Um eine neue Version einer Datei zu erzeugen, muss diese ersetzt werden. Dies kann auch für große Dateien mit dem unten beschriebenen Map-Reduce Algorithmus sehr effektiv durchgeführt werden.

Im Kontext eines Hadoop-Systems können nun beispielsweise im CSV-Format abgelegte Textdateien mit Map-Reduce Jobs verarbeitet werden, wobei die Verteilung der Teil-Jobs über den Cluster basierend auf der Verteilung der HDFS-Dateiblöcke automatisch vom Hadoop-Framework übernommen wird. Neben reinen Textdateien können auch strukturierte Binärdaten wie zum Beispiel die Formate ORC, Parquet, oder AVRO direkt von Hadoop verarbeitet werden. Für neue Formate können darüber hinaus spezifische Splitter-Klassen für Map-Reduce implementiert werden.

Weiters ist es im Rahmen einer Map-Reduce Algorithmus ohne Probleme möglich, beispielsweise nur eine Map-Stufe durchzuführen, um neue Spalten zu einer CSV-Datei hinzuzufügen.

Map-Reduce Framework (YARN)

Basierend auf der oben gezeigten Datenverteilung in HDFS kann nun durch den Dienst Yarn ein datenparalleler Batch-Job ausgeführt werden, wobei jeder Slaveknoten die lokal vorliegenden Datenblöcke bearbeitet. Konzeptionell folgt die Ausführung dabei dem Map-Reduce Algorithmus. Ein klassisches Anwendungsbeispiel für den Map-Reduce Algorithmus ist das Zählen von Wörtern in Textdokumenten. Hier emitiert der Map-Schritt pro Zeile eine Menge an Paaren der Form (Wort, Anzahl des Auftretens in der Zeile). Im Shuffle-Schritt werden diese Wertpaare nach den Worten zusammengefasst, da diese den Schlüssel repräsentieren. Im abschliessenden Reduce-Schritt werden die Worthäufigkeiten pro Wort aufsummiert. Eine beispielhafte Ausführung könnte wie folgt ablaufen:

1. Der Eingabetext wird in einzelne Textzeilen aufgeteilt. (Splitting)
2. Der Map-Schritt, welcher für jede Zeile einzeln parallel ausgeführt wird, erzeugt für jedes Wort in der Zeile ein Paar aus dem Wort und der Zahl 1. (Mapping)
3. Die Paare werden nach den Worten sortiert und zu einer Liste je Wort zusammengefasst. (Shuffling)
4. Für jedes Wort wird durch Aufsummieren der Zahlen die Anzahl im Gesamttext bestimmt. (Reducing)

Während der Map-Schritt und der Reduce-Schritt jeweils ausprogrammiert werden müssen, wird der globale Shuffle-Schritt automatisch vom Map-Reduce Framework übernommen. Die Implementierung des Map- und Reduce-Schritts erfordert in der Praxis beispielsweise das objektorientierte Überschreiben von jeweils einer Map- und einer Reduce-Methode, deren Schnittstellen bereits vorgegeben sind. Dies ermöglicht den Fokus auf die Transformation eines Wertepaares zu setzen, während sich das Framework in weiterer Folge um die skalierte Ausführung am Cluster kümmert.

Batch- und Datenstromverarbeitung: Apache Spark

Spark stellt eine flexible Datenverarbeitungschicht dar, die auf verschiedene Infrastrukturen wie beispielsweise Hadoop aufgesetzt und für verschiedene Aufgaben im Bereich von Data Engineering aber auch Data Science genutzt werden kann. Als allgemeines Datenverarbeitungsframework kann Spark Aufgaben der Datenvorverarbeitung genauso wie Machine Learning Aufgaben übernehmen.

Es kann beispielsweise auf einem bestehenden Hadoop-Cluster installiert werden und direkt auf die dort gespeicherten Daten zugreifen und diese parallel verarbeiten. Ein Ansatz dazu ist der oben genannte Map-Reduce Algorithmus, wobei allerdings Spark noch andere flexible Methoden wie beispielsweise Datenfilterung anwenden kann. Spark legt Zwischenergebnisse als verteilte Datensets (resilient distributed datasets, RDDs) im Hauptspeicher ab, wodurch langsame wiederholte Festplattenzugriffe – wie häufig bei klassischen Datenbanken – vermieden werden können.

Wichtige Features von Spark umfassen unter anderem:

• Datenparallele Batchverarbeitung zum Beispiel unter Einsatz des Map-Reduce Algorithmus.
• Unterstützung von SQL-Abfragen auf beliebig (z.B. im HDFS) gespeicherte Daten. Hierzu muss nur interaktiv eine Tabelle angelegt werden, die das zu verwendende Datenschema definiert und auf die zugrundeliegenden Daten verweist.
• Basierend auf der seqentielle Verarbeitung von mehreren RDDs kann Datenstromverarbeitung durchgeführt werden.

Genauso wie ein zugrundeliegender Hadoop-Cluster kann eine Spark-Installation durch eine einfache Hardware-Erweiterung für die Verarbeitung größerer Datenmengen fit gemacht werden.

Anwendungsbeispiel: Aufbereitung von industriellen Sensor- und Logdaten

Im Rahmen des Forschungsprojekts VPA4.0 wurde in eine Datenpipeline zur Vorverabeitung der Produktions-Sensordaten eingerichtet. Diese stellt ein gutes Beispiel für die Verknüpfung von Streaming- und Batch-Verarbeitung dar. Apache NiFi wurde als Streaming-Lösung verwendet, um die Daten direkt vom Projektpartner verschlüsselt über das Internet zu übertragen, bevor sie lokal am Hadoop-Cluster gespeichert wurden. Die weitere Datenaufbereitung wurde dann mithilfe von Spark parallel auf dem Hadoop-Cluster durchgeführt und umfasste folgende Schritte:

1. Entpacken der erhaltenen Datenarchive und Entfernung nicht benötigter Dateien
2. Aufbereitung und Speicherung der Daten als CSV-Textdateien im HDFS
3. Anlegen virtueller Tabellen basierend auf CSV Dateien ermöglicht weitere Verarbeitung mit SQL
4. Datenfilterung und Speicherung im optimierten Parquet-Format für interaktive SQL-Abfragen

Umsetzung in Spark: >> siehe ganzen Fachbeitrag der RISC Software GmbH

Umsetzung in NiFi:

Das hier gezeigte Beispiel zeigt wiederum das Einlesen der CSV-Datei mit den Wärmekraftmaschinendaten und deren Speicherung in einer SQLite-Datenbank. Hier wird die CSV-Datei mithilfe des GetFile-Prozessors eingelesen und in NiFi-Flowfiles umgewandelt. Diese werden in einen PutDatabaseRecord-Prozessor geleitet, der für das korrekte Parsing der CSV-Datei und den Zugriff auf die Datenbank konfiguriert wird. Genauso wie das Verbinden der einzelnen Module erfolgt auch deren Konfiguration interaktiv in der NiFi-Weboberfläche.

Der abschließende PutFile-Prozessor dient zum Abfangen und Speichern von Fehlerbedingungen, wie beispielsweise falsch formatierte Zeilen in der Eingabedatei. Dadurch wird ermöglicht, das Fehlerbedingungen leicht in der gespeicherten Textdatei nachvollzogen werden können.

Die Wahl der passenden Werkzeuge

Wie aus den gezeigten Beispielen der Datenüberführung aus einer CSV-Datei in eine SQL-Datenbank ersichtlich ist, führen im Bereich Data Engineering oft verschiedene Wege zum selben Ziel. Welche Methoden zum Einsatz kommen sollten, hängt oft von den speziellen Anforderungen der Kunden sowie deren Systemumgebung ab:

• Wenn beispielsweise bereits ein Hadoop-Cluster im Einsatz oder geplant ist, kann dieser beim Entwurf einer Lösung bereits eingebunden werden.
• Public-Cloud Angebote wie beispielsweise Amazon AWS bieten wiederum Alternativen zu den oben beschriebenen Open-Source Lösungen an, welche vor allem den Betrieb der Lösung vereinfachen, aber auch zu Vender-Lock-in führen können.
• Weitere Kriterien für eine Technologieentscheidung sind Anforderungen an Skalierbarkeit und die geplante Integration zusätzlicher Werkzeuge
• Nicht zuletzt bieten Open-Source Lösungen oft Kostenvorteile, da auch für hoch skalierende Lösungen keine Lizenzkosten anfallen.

Autor

DI Paul Heinzlreiter ist Senior Data Engineer in der Abteilung Logistics Informatics der RISC Software GmbH.

>> Fachbeitrag der RISC Software GmbH

Kontakt

RISC Software GmbH
Softwarepark 32a
4232 Hagenberg
www.risc-software.at