Grundlagen-Tutorial

Als Erstes werden wir einen InvokeHTTP Processor anlegen.
Der InvokeHTTP Processor kann beliebige HTTP-Requests ausführen. Mit einem GET-Request auf die entsprechende URL laden wir die CSV-Datei herunter. Die Datei wird intern in ein FlowFile übersetzt und kann an weitere Processors weitergeleitet werden.

In diesem Abschnitt werden wir einen Processor anlegen, der die CSV-Datei entgegennimmt und nach neuen Bahnhöfen filtert.
Wir wissen aus der Testausführung, in welchem Format die Daten vorliegen und müssen den nachfolgenden Processor für die Verarbeitung bestimmen.

Einer der Processors, den wir für unser Anliegen verwenden können, ist der QueryRecord Processor.

Der QueryRecord Processor selektiert über SQL-Queries den Content eingehender FlowFiles. Dazu müssen die Daten in einem Format vorliegen, dass NiFi als Records interpretieren kann - das CSV-Format gehört dazu.

Über das SQL kann man dann filtern, transformieren oder aggregieren.

Die Ergebnisse der Queries werden als FlowFiles entlang einer selbstdefinierten Relation weitergeschickt.

Zuerst werden wir den Processor ins Diagramm einfügen und ihn mit der Response Relation des InvokeHTTP Processors verbinden.

Als Nächstes müssen wir den QueryRecord Processor konfigurieren, damit er mit den eingehenden Daten wie gewünscht umgeht.

Unser Ziel für Teil 1 ist es, ein komprimiertes XML mit Informationen über alle neuen Bahnhöfe abzuspeichern. Diese komprimierte XML könnten wir dann als Datei über verschiedene Processors auf externe Systeme übertragen. Zum Beispiel könnten wir mit dem PutEmail Processor die Datei als Anhang mitgeben oder aber mit PutSFTP auf einem Fileserver ablegen. In vielen Fällen übergeben wir dabei den Content als Datei.

Wir können den Namen dieser Datei bestimmen, in dem wir das Attribut “filename” setzen, das von den Processors dann ausgelesen wird. Auch, wenn wir in diesem Tutorial den Content nur per IGUASU-Benutzeroberfläche herunterladen, wird auch der Name der Download-Datei durch dieses Attribut gesetzt.
Mit dem UpdateAttribute Processor können wir beliebige Attribute der FlowFiles anlegen und ändern.

Wir fügen einen UpdateAttribute Processor zum Diagramm hinzu und verbinden diesen mit der "neue_alle"-Relation aus dem QueryRecord Processor.

Das Anlegen neuer Attribute über den UpdateAttribute Processor erfolgt durch das Erstellen einer dynamischen Property.
Der Name dieses dynamischen Properties entspricht dann dem Namen des Attributs, das angelegt (oder geändert) werden soll. Der Inhalt entspricht dann dem (neuen) Wert.

Den Inhalt möchten wir auf das aktuelle Datum setzen. Da sich das Datum täglich ändern wird, können wir den String nicht statisch eintragen, sondern müssen diesen dynamisch passend zum Tag generieren.

Dafür nutzen wir die Expression Language, mit der wir dynamische Funktionen in Ausdrücke für Properties eintragen können.

Die Expression ${now():format('yyyy-MM-dd')} würde ein aktuelles Datum wie "2020-11-19" ergeben. Da wir die Datei als XML abspeichern wollen, können wir an die Expression noch das statische ".xml" anhängen und mit einem Prefix "Stations_" versehen: Stations_${now():format('yyyy-MM-dd')}.xml.
Das ergibt dann "Stations_2020-11-19.xml".

Der UpdateAttribute Processor ändert nur die definierten Attribute des FlowFiles, nicht den Content. Das resultierende FlowFile mit dem nun angepassten "filename" werden wir an den nächsten Processor weiterleiten.

In diesem Schritt komprimieren wir das FlowFile. Dazu setzen wir den CompressContent Processor ein. Wir müssen den Processor zum Diagramm hinzufügen und nur noch die Relation aus dem UpdateAttribute anlegen.

In den Einstellungen sollten wir:

Das komprimierte FlowFile leiten wir an den letzten Processor weiter.

Wir können uns vorstellen, dass die komprimierte XML-Datei im nächsten Schritt an ein externes System weitergeleitet oder aber zur Sicherung abgespeichert wird. Wir simulieren diesen Schritt, indem wir die Analyse Tools von IGUASU nutzen, um den Content herunterzuladen, die Datei zu entpacken und schlussendlich das XML zu betrachten.
Dafür lassen wir die FlowFiles in ein Funnel-Element laufen, um sie später zu betrachten.

Ein Funnel ist ein einfaches Element in IGUASU. Es dient dazu, FlowFiles aus einer oder mehreren eingehenden Verbindungen zu sammeln, beendet aber den Datenfluss an dieser Stelle. Das ist nützlich, um Zweige eines Flows abzuschließen oder um FlowFiles gezielt für eine manuelle Inspektion über die Benutzeroberfläche 'aufzufangen', ohne dass ein weiterer Processor sie verarbeitet.

Durch einen Rechtsklick ins Diagramm und die Auswahl Start können wir alle Processors der Process Group starten. Wenn wir alles richtig gemacht haben, sollten wir nach kurzer Zeit sehen, dass ein FlowFile unsere Processors durchlaufen hat und nun vor dem Funnel wartet.

Wir klicken auf dieses wartende Flowfile, um auf der rechten Seite die FlowFile und eine Übersicht der Queue auf dieser Relation zu sehen.

Über das -Icon im Content-Bereich können wir den Inhalt des Contents herunterladen.

Wenn wir die Datei herunterladen, sollten wir sehen, dass der Name “Stations_[yyyy-MM-dd].xml.gz” ist - wie vorher von uns bestimmt. Die Dateiendung “.gz” weist auf eine mit gzip komprimierte Datei hin. Wenn wir die Datei entpacken und die daraus resultierende XML in einem Editor öffnen, können wir nun die neuen Bahnhöfe als XML betrachten.

Zunächst benötigen wir einen Processor, der die HTTP Anfragen empfängt und weiterleitet. Dafür eignet sich der HandleHttpRequest Processor.
In der Konfiguration setzten wir den Listening Port auf "3333". Das ist der Port, an dem der Processor auf Anfragen horchen wird.

Außerdem müssen wir einen neuen StandardHttpContextMap Controller Service anlegen, der dafür sorgt, dass die HTTP Session zwischen den Processors erhalten bleibt.

Wir legen, wie aus Teil 1 bekannt, einen neuen StandardHttpContextMap Controller Service an, diesmal jedoch innerhalb der aktuellen Process Group Validierung via REST.
Wir könnten den Controller Service auch in der Parent Process Group “Deutsche Bahn Tutorial” oder gar in der Root Process Group anlegen, da die Controller Services einer Process Group auch allen Sub-Process Groups zur Verfügung stehen. Jedoch vermeiden wir es hier, da wir den Controller Service ausschließlich in dieser Process Group verwenden wollen.

Es bietet sich an, bei der Erstellung dieses Controller Services den Namen mit einem Prefix "REST" zu versehen, um Verwechslung mit einem anderen StandardHttpContextMap Controller Services zu vermeiden.

Die optionale Property Allowed Paths befüllen wir mit “/validate”.
Das hat zur Folge, dass der Processor nur Requests mit diesem Pfad in der URL entgegennimmt. Alle Requests, die nicht an localhost:3333/validate gehen, werden automatisch mit einem Code 404 (Not Found) beantwortet.

Weil wir einen REST Service anbieten, an den Daten zur Überprüfung geschickt werden sollen, sollte der Client beim Senden der Daten die “POST” HTTP Methode benutzen. Daher entfernen wir die Häkchen an den Einstellungen “Allow GET”, “Allow PUT” und “Allow DELETE”, sodass nur noch “Allow POST” aktiviert ist.

Um die Anfragen zu beantworten, benötigen wir außerdem zwei weitere HandleHttpResponse Processors.
Der erste soll bei erfolgreicher Validierung dem Client einen Code 200 als Erfolgsmitteilung zurückgeben. Der andere wiederum soll bei Auftreten eines Fehlers dies dem Client mit einem Code 400 mitteilen.

In einem realen Szenario wäre es angemessen, die unterschiedlichen Fehlerfälle (keine valide JSON übergeben, JSON Schema Validierung fehlgeschlagen) auch unterschiedlich zu beantworten und auch weitere Informationen zum Fehler zu liefern. Für dieses Tutorial beschränken wir uns auf die Response mit dem Code 400 für beide Fehlerfälle.

Wir erstellen zwei HandleHttpResponse Processors und setzen in beiden Fällen unseren "REST StandardHttpContextMap" Controller Service in das Feld HTTP Context Map.

Das Feld HTTP Status Code setzen wir entsprechend auf 200 beziehungsweise 400.

Im nächsten Schritt wollen wir die erhaltenen JSON-Daten auf das richtige Format überprüfen. Dafür eignet sich der ValidateJson Processor, der JSONs anhand von JSON Schema validieren kann.

Ein JSON-Schema ist dabei selbst ein JSON, das Vorgaben zum Aufbau und Inhalt eines anderen JSON macht, also ein Schema definiert. Der ValidateJson Processor prüft, ob unser JSON die Vorgaben des JSON-Schemas erfüllt oder nicht.

Mit dem folgenden JSON-Schema definieren wir ein JSON Objekt, das einen Eintrag zu einem Bahnhof repräsentiert. Es hat genau drei Einträge:

Dieses Schema muss in das "JSON Schema" Property eingetragen werden. Ansonsten können die Standardeinstellungen des Processors beibehalten werden.

Die resultierende Konfiguration des ValidateJson Processors sollte schlussendlich so aussehen:

Die validen Einträge werden entlang der "valid"-Relation zum HandleHttpResponse Processor mit der erfolgreichen Response Code 200 geschickt, während im Falle eines Validierungsfehlers der andere HandleHttpResponse Processor den Response Code 400 zurückschickt.

Am Ende sollte sich folgender Flow ergeben:

Unser als untergeordnete Process Group implementierter REST Service zum Überprüfen der Daten ist nun fertig und kann gestartet werden.

In diesem Teil wollen wir aus den CSV-Daten der Bahnhöfe zwecks einer besseren Übersicht nur die Bahnhofsnummer (EVA_NR als NR) und den Bahnhofsnamen (NAME) behalten. Wir können den vorher erstellten QueryRecord Processor benutzen, um alle anderen Informationen zu entfernen und gleichzeitig, wie auch zuvor, die Daten in ein XML-Format übertragen.
Dazu erstellen wir eine neue Dynamic Property mit dem Namen “neue” und setzen als Wert das passende SQL Statement.

Sowohl “neue” als auch “neue_alle” sind jetzt Ausgänge des QueryRecord Processors. Um die Funktion zu überprüfen, können wir eine Testausführung mit den vorhandenen CSV-Daten machen (siehe Isolierte Prozessausführung (Fortführung)).

Wir wollen, dass sowohl Teil 1 als auch Teil 2 bei jeder Ausführung des Flows durchlaufen werden. Daher werden wir Teil 1 nicht verändern, sondern lassen beide Teile parallel existieren und laufen.

Hierzu verbinden wir unseren nächsten neuen Processor, den UpdateRecord Processor, mit dem gerade angelegten “neue”-Ausgang des QueryRecord Processors. Dieser gibt somit an zwei verschiedene Ausgänge zwei unterschiedliche FlowFiles weiter, die unabhängig voneinander behandelt werden.

Wir wollen jetzt, neben der Nummer und dem Namen des Bahnhofs, zu jedem Eintrag noch das aktuelle Datum hinzufügen. Damit wollen wir zeigen, dass die Informationen dem heutigen Standpunkt entsprechen. Wir verwenden hierfür den UpdateRecord Processor.

Wie beim QueryRecord Processor in Teil 1, müssen wir einen Record Reader Controller Service und einen Record Writer Controller Service spezifizieren. Da wir unsere Daten im XML-Format von QueryRecord bekommen, brauchen wir einen XMLReader.

Diesen müssen wir, wie im ersten Teil, als Service zur Process Group hinzufügen und in der Konfiguration die Einstellung Expect Records as Array auf "true" setzen, da die Daten als Array im XML ankommen (siehe "Result content" unter Testausführung des QueryRecord Processors).

Für den Record Writer werden wir einen JsonRecordSetWriter Processor mit Standardeinstellungen verwenden. Damit werden die Stationen aus dem XML- in das JSON-Format umgewandelt, da wir letztendlich ein JSON an die REST-Schnittstelle schicken wollen.

Anschließend fügen wir ein Dynamic Property über den -Button hinzu. Wir nutzen wieder die Expression für das aktuelle Datum und fügen dieses damit zu jedem Record hinzu. Der Name des Dynamic Property gibt dabei den Namen des zusätzlichen Feldes im Record vor.
(Dieses Vorgehen ähnelt dem vorhergehenden Schritt mit dem UpdateAttribute Processor im ersten Teil des Tutorials, siehe 1.3 UpdateAttribute Processor (Dateinamen setzen).)

Bei der Ausführung werden die Stationen jeweils aus XML in JSON umgewandelt und mit dem aktuellen Datum hinter dem Namen "CREATED" versehen:

Die FlowFile mit der JSON leiten wir nun in einen SplitRecord Processor weiter. Mit diesem Processor werden wir die Liste der Bahnhöfe in einzelne Elemente aufteilen, sodass pro Bahnhof ein Aufruf des REST-Services erfolgen kann.

Da die Daten im JSON-Format ankommen, brauchen wir einen JSONTreeReader als Record Reader, den wir wieder als Controller Service erzeugen müssen.
Die Standardeinstellungen können beibehalten werden.

Für den Record Writer können wir den JsonRecordSetWriter verwenden, den wir vorher für den UpdateRecord Processor angelegt haben.

Jedes FlowFile aus dem SplitRecord Processor wird jetzt noch an einen weiteren InvokeHTTP Processor weitergeleitet. Mit diesem Processor rufen wir den REST Service zur Überprüfung der JSON-Daten auf und übergeben den FlowFile Content, also das JSON mit dem Eintrag zu einem Bahnhof, als Body.

Der Flow sieht nun wie folgt aus:

Durch einen Rechtsklick ins Diagramm und die Auswahl Start starten wir alle Processors.

Der obere InvokeHTTP Processor sendet eine HTTP-Anfrage zum Erhalt der CSV-Datei. Sobald er diese mit der Response erhält, wird das FlowFile weitergereicht an den QueryRecord Processor.

Hier spaltet sich der Flow in zwei Branches.
Links wird über die “neue_alle” Relation eine XML aller neuen Bahnhöfe mit allen vorhandenen Infos an den UpdateAttribute Processor übergeben. Mit dem UpdateAttribute Processor wird das Attribute “filename” gesetzt, das den Namen der später ausgegebenen Datei bestimmt.

Die Content und Attributes umfassende FlowFile wird weitergereicht an den CompressContent Processor, wo aus der XML ein mit gzip komprimierter Datenstrom wird. Dieser landet schlussendlich in der Relation zum Funnel, wo wir uns den Content anschauen und herunterladen können.

Auf dem anderen Branch des Flows wird vom QueryRecord Processor über die “neue”-Relation eine XML ebenfalls mit Daten zu den neuen Bahnhöfen, jedoch nur den Namen und die Nummer, übergeben. Der UpdateRecord Processor fügt dann jedem dieser Einträge eine weitere Information hinzu, nämlich das heutige Datum.

Gleichzeitig konvertiert er die Daten in das vom REST-Endpunkt erwartete JSON-Format. Dieses JSON mit Einträgen zu Bahnhöfen geht dann in den SplitRecord Processor, wo jeder Eintrag einzeln in ein FlowFile gepackt wird, sodass nun statt einem mehrere FlowFiles den Processor verlassen.

Jedes dieser FlowFiles läuft einzeln über die “splits” Response in den InvokeHTTP Processor, der für jede der FlowFiles eine Request mit dem JSON aus dem Content an den von uns implementierten REST Service sendet.

Jedes FlowFile mit validen JSON-Daten produziert eine Response des REST Service mit Status Code 200 und wird dementsprechend in die Relation “Response” geleitet.