Apache NIFI – Ein kurzer Überblick über die praktischen Möglichkeiten

Einführung

Es stellte sich heraus, dass ich an meinem aktuellen Arbeitsplatz mit dieser Technologie in Kontakt kam. Lassen Sie mich mit einer kurzen Vorgeschichte beginnen. Bei einem unserer Meetings wurde unserem Team mitgeteilt, dass wir eine Integration mit einem bekannten Systemerstellen müssen. Unter dieser Integration verstand man, dass dieses bekannte System uns Anfragen über HTTP an einen bestimmten Endpunkt senden würde, und wir, wie seltsam das auch erscheinen mag, würden im Gegenzug Antworten in Form von SOAP-Nachrichten zurücksenden. Das scheint alles einfach und trivial zu sein. Daraus folgt, dass wir…

Ziel

Erstellen Sie 3 Dienste. Der erste davon ist der Datenbankaktualisierungsdienst. Dieser Dienst aktualisiert die Daten in der Datenbank, sobald neue Daten von einem externen System eintreffen, und generiert eine Datei im CSV-Format zur Übertragung an das nächste System. Er ruft den Endpunkt des zweiten Dienstes – des Transportdienstes über FTP – auf, der die übertragene Datei erhält, sie validiert und in den Dateispeicher über FTP ablegt. Der dritte Dienst – der Datenübertragungsdienst an den Endkunden – arbeitet asynchron zu den ersten beiden. Er akzeptiert eine Anfrage von einem externen System für die oben genannte Datei, holt die fertige Antwortdatei, modifiziert sie (aktualisiert die Felder id, description, linkToFile) und sendet die Antwort als SOAP-Nachricht. Insgesamt sieht die Situation folgendermaßen aus: Die ersten beiden Dienste beginnen ihre Arbeit nur, wenn Daten zur Aktualisierung eintreffen. Der dritte Dienst arbeitet ständig, da es viele Informationsanfragen gibt, etwa 1000 Anfragen pro Minute. Die Dienste sind durchgehend verfügbar, und ihre Instanzen befinden sich in verschiedenen Umgebungen wie Test-, Demo-, Vorproduktions- und Produktionsumgebungen. Unten ist das Schema der Funktionsweise dieser Dienste dargestellt. Ich möchte gleich erklären, dass einige Details vereinfacht wurden, um überflüssige Komplexität zu vermeiden.

Technische Vertiefung

Bei der Planung der Lösung haben wir zunächst beschlossen, eine Anwendung in Java unter Verwendung des Spring Frameworks zu entwickeln, unterstützt von einem Nginx-Load-Balancer und einer Postgres-Datenbank, sowie weiteren technischen und weniger technischen Elementen. Da der Zeitrahmen für die technische Entwicklung die Untersuchung alternativer Ansätze erlaubte, fielen unsere Blicke auf die in bestimmten Kreisen modische Technologie Apache NIFI. Um es gleich vorweg zu sagen: Diese Technologie hat es uns ermöglicht, diese drei Dienste zu erkennen. In diesem Artikel wird die Entwicklung eines Dateiübertragungsdienstes und eines Datenübertragungsdienstes an den Endnutzer beschrieben. Sollte das Interesse bestehen, werde ich auch über den Dienst zur Aktualisierung von Daten in der Datenbank schreiben.

Was ist das?

NIFI ist eine verteilte Architektur, die eine schnelle parallele Datenaufnahme und -verarbeitung ermöglicht. Sie bietet eine Vielzahl von Plugins für Datenquellen und Transformationen, Konfigurationsversionierung und vieles mehr. Ein zusätzlicher Vorteil ist die Benutzerfreundlichkeit. Triviale Prozesse wie getFile, sendHttpRequest und andere können als Kästchen dargestellt werden. Jedes Kästchen repräsentiert einen bestimmten Prozess, dessen Interaktionen im folgenden Bild zu sehen sind. Eine detaillierte Dokumentation über die Interaktion und Konfiguration von Prozessen ist geschrieben. hier , für diejenigen, die auf Russisch sind — hier. In der Dokumentation wird hervorragend erklärt, wie man NIFI entpackt und startet sowie wie man Prozesse erstellt, die ebenfalls als Kästchen dargestellt werden.
Die Idee, diesen Artikel zu schreiben, entstand nach längerer Suche und Strukturierung der gesammelten Informationen in etwas Verständliches und dem Wunsch, zukünftigen Entwicklern das Leben etwas zu erleichtern.

Beispiel

Hier wird ein Beispiel vorgestellt, wie Quadrate miteinander interagieren. Das allgemeine Schema ist ziemlich einfach: Wir erhalten eine HTTP-Anfrage (theoretisch mit einer Datei im Anfragekörper. Zur Demonstration der Möglichkeiten von NIFI startet in diesem Beispiel die Anfrage den Prozess, eine Datei aus dem lokalen FH zu erhalten), daraufhin wird eine Antwort gesendet, dass die Anfrage eingegangen ist, während parallel der Prozess zum Abrufen der Datei aus dem FH gestartet wird und anschließend der Prozess für die Übertragung über FTP ins FH. Es sei erwähnt, dass die Prozesse über das sogenannte FlowFile miteinander kommunizieren. Dies ist eine grundlegende Entität in NIFI, die Attribute und Inhalt speichert. Der Inhalt sind die Daten, die durch die Stream-Datei dargestellt werden. Grob gesagt, wenn Sie eine Datei aus einem Quadrat erhalten und sie an ein anderes weitergeben, ist der Inhalt Ihre Datei.

Wie Sie an diesem Bild erkennen können, zeigt es den allgemeinen Prozess. HandleHttpRequest – empfängt Anfragen, ReplaceText – generiert den Antwortinhalt, HandleHttpResponse – gibt die Antwort zurück. FetchFile – erhält eine Datei aus dem Dateispeicher und überträgt sie an das Quadrat PutSftp – legt diese Datei an dem angegebenen FTP-Speicherort ab. Jetzt erfahren wir mehr über diesen Prozess.

In diesem Fall ist der Request der Anfang für alles. Lassen Sie uns seine Konfigurationseinstellungen betrachten.

Hier ist alles ziemlich einfach, mit der Ausnahme von StandartHttpContextMap – das ist ein Service, der es ermöglicht, Anfragen zu senden und zu empfangen. Weitere Informationen und sogar Beispiele finden Sie hier – hier

Als Nächstes sehen wir uns die Konfigurationseinstellungen von ReplaceText an. Hierbei sollten Sie auf ReplacementValue achten – das ist das, was dem Benutzer als Antwort zurückgegeben wird. In den Einstellungen können Sie das Logging-Level anpassen; die Protokolle finden Sie unter {куда распаковали nifi}/nifi-1.9.2/logs. Dort gibt es auch die Parameter failure/success – basierend auf diesen Parametern können Sie den gesamten Prozess steuern. Das heißt, im Falle einer erfolgreichen Textverarbeitung wird der Prozess des Sendens einer Antwort an den Benutzer ausgelöst, während wir im anderen Fall einfach den fehlgeschlagenen Prozess protokollieren.

In den Eigenschaften von HandleHttpResponse gibt es außer dem Status bei erfolgreicher Erstellung der Antwort nicht viel Interessantes.

Nun, da wir den Request und die Antwort bearbeitet haben, gehen wir weiter zum Abrufen der Datei und deren Platzierung auf dem FTP-Server. FetchFile – holt eine Datei aus dem in den Einstellungen angegebenen Pfad und übergibt sie an den nächsten Prozess.

Der PutSftp-Block lädt die Datei in den Speicher. Die Konfigurationsparameter sind unten aufgeführt.

Es ist wichtig zu beachten, dass jeder Block ein separater Prozess ist, der gestartet werden muss. Wir haben das einfachste Beispiel betrachtet, das keine komplizierte Anpassung erfordert. Nun schauen wir uns einen etwas komplexeren Prozess an, bei dem wir ein wenig mit Groovy programmieren.

Ein komplexeres Beispiel

Der Datenübertragungsdienst an den Verbraucher ist etwas komplizierter, da der Prozess der Modifikation von SOAP-Nachrichten hinzugefügt wurde. Der gesamte Prozess ist im folgenden Diagramm dargestellt.

Die Idee ist auch hier nicht besonders kompliziert: Wir haben die Anfrage vom Verbraucher erhalten, dass er Daten benötigt, haben eine Antwort gesendet, dass wir die Nachricht erhalten haben, den Prozess zur Datenbeschaffung gestartet, diesen mit einer bestimmten Logik bearbeitet und dann die Datei in Form einer SOAP-Nachricht an den Server übergeben.

Ich denke, es ist nicht nötig, die bereits oben erwähnten Blöcke erneut zu beschreiben – lassen Sie uns sofort zu den neuen übergehen. Wenn Sie eine Datei bearbeiten müssen und herkömmliche Blöcke wie ReplaceText nicht geeignet sind, müssen Sie Ihr eigenes Skript schreiben. Dies kann mithilfe des Blocks ExecuteGroogyScript erfolgen. Die Einstellungen dazu sind unten aufgeführt.

Es gibt zwei Möglichkeiten, ein Skript in dieses Quadrat zu laden. Die erste besteht darin, die Datei mit dem Skript hochzuladen. Die zweite Möglichkeit ist, das Skript in den scriptBody einzufügen. Soweit ich weiß, unterstützt das Quadrat executeScript mehrere Programmiersprachen – eine davon ist Groovy. Leider müssen Java-Entwickler enttäuscht werden: In einem solchen Quadrat kann man keine Skripte in Java schreiben. Für diejenigen, die es unbedingt möchten, muss ein benutzerdefiniertes Quadrat erstellt und in das NIFI-System integriert werden. Dieser gesamte Vorgang erfordert leider einiges an Aufwand, mit dem wir uns im Rahmen dieses Artikels jedoch nicht befassen werden. Ich habe mich für die Sprache Groovy entschieden. Im Folgenden finden Sie ein Testskript, das einfach inkrementell die ID in der SOAP-Nachricht aktualisiert. Wichtig zu beachten: Sie nehmen die Datei aus flowFile, aktualisieren sie und sollten nicht vergessen, die aktualisierte Version wieder zurückzulegen. Ebenfalls zu beachten ist, dass nicht alle Bibliotheken eingebunden sind. Es kann also sein, dass Sie tatsächlich eine der Bibliotheken importieren müssen. Ein weiterer Nachteil ist, dass das Debugging des Skripts in diesem Quadrat ziemlich schwierig ist. Es besteht die Möglichkeit, sich mit der JVM von NIFI zu verbinden und den Debugging-Prozess zu starten. Ich habe persönlich eine lokale Anwendung gestartet und den Empfang der Datei aus der Sitzung simuliert. Das Debugging wurde ebenfalls lokal durchgeführt. Die Fehler, die beim Laden des Skripts auftreten, lassen sich relativ leicht googeln und werden von NIFI im Log verzeichnet.

import org.apache.commons.io.IOUtils
import groovy.xml.XmlUtil
import java.nio.charset.*
import groovy.xml.StreamingMarkupBuilder

def flowFile = session.get()
if (!flowFile) return
try {
    flowFile = session.write(flowFile, { inputStream, outputStream ->
        String result = IOUtils.toString(inputStream, "UTF-8");
        def recordIn = new XmlSlurper().parseText(result)
        def element = recordIn.depthFirst().find {
            it.name() == 'id'
        }

        def newId = Integer.parseInt(element.toString()) + 1
        def recordOut = new XmlSlurper().parseText(result)
        recordOut.Body.ClientMessage.RequestMessage.RequestContent.content.MessagePrimaryContent.ResponseBody.id = newId

        def res = new StreamingMarkupBuilder().bind { mkp.yield recordOut }.toString()
        outputStream.write(res.getBytes(StandardCharsets.UTF_8))
} as StreamCallback)
     session.transfer(flowFile, REL_SUCCESS)
}
catch(Exception e) {
    log.error("Fehler bei der Verarbeitung von validate.groovy", e)
    session.transfer(flowFile, REL_FAILURE)
}

An dieser Stelle endet die Anpassung des Blocks. Anschließend wird die aktualisierte Datei an den Block übergeben, der für den Versand der Datei an den Server zuständig ist. Nachfolgend sind die Einstellungen dieses Blocks aufgeführt.

Wir beschreiben die Methode, mit der die SOAP-Nachricht übermittelt wird. Wir schreiben, wohin. Anschließend muss angegeben werden, dass es sich tatsächlich um SOAP handelt.

Wir fügen mehrere Eigenschaften wie Host und Aktion (soapAction) hinzu. Wir speichern und prüfen. Genauere Informationen zum Versenden von SOAP-Anfragen finden Sie hier. hier

Wir haben mehrere Anwendungsfälle für NIFI-Prozesse untersucht. Wie sie interagieren und welchen echten Nutzen sie bringen. Die betrachteten Beispiele sind Testfälle und weichen ein wenig von den realen Einsätzen ab. Ich hoffe, dieser Artikel wird für Entwickler etwas nützlich sein. Vielen Dank für Ihre Aufmerksamkeit. Bei Fragen können Sie mir gerne schreiben. Ich werde versuchen, zu antworten.

Quelle: habr.com