Apache NIFI – Ein kurzer Überblick über Möglichkeiten in der Praxis

Einführung

So kam es, dass ich mich an meinem jetzigen Arbeitsplatz mit dieser Technologie vertraut machen musste. Ich beginne mit einem kleinen Hintergrund. Beim nächsten Treffen wurde unserem Team mitgeteilt, dass wir eine Integration schaffen müssen bekanntes System. Mit Integration war gemeint, dass dieses bekannte System uns Anfragen über HTTP an einen bestimmten Endpunkt sendet und wir seltsamerweise Antworten in Form einer SOAP-Nachricht zurücksenden. Alles scheint einfach und trivial. Daraus folgt, dass Sie...

Aufgabe

Erstellen Sie 3 Dienste. Der erste davon ist der Datenbankaktualisierungsdienst. Wenn neue Daten von einem Drittsystem eintreffen, aktualisiert dieser Dienst die Daten in der Datenbank und generiert eine Datei im CSV-Format, um sie an das nächste System zu übertragen. Der Endpunkt des zweiten Dienstes heißt FTP-Transportdienst, der die übertragene Datei empfängt, validiert und per FTP im Dateispeicher ablegt. Der dritte Dienst, der Consumer-Datenübertragungsdienst, arbeitet asynchron mit den ersten beiden. Es empfängt von einem externen System eines Drittanbieters eine Anfrage zum Empfang der oben beschriebenen Datei, nimmt die fertige Antwortdatei, ändert sie (aktualisiert die Felder „ID“, „Beschreibung“ und „LinkToFile“) und sendet die Antwort in Form einer SOAP-Nachricht. Das heißt, das Gesamtbild ist wie folgt: Die ersten beiden Dienste beginnen ihre Arbeit erst, wenn die Daten zur Aktualisierung eingetroffen sind. Der dritte Dienst funktioniert ständig, da es viele Informationskonsumenten gibt, etwa 1000 Datenanfragen pro Minute. Dienste sind ständig verfügbar und ihre Instanzen befinden sich in verschiedenen Umgebungen, z. B. Test-, Demo-, Vorproduktions- und Produktionsumgebungen. Nachfolgend finden Sie ein Diagramm zur Funktionsweise dieser Dienste. Lassen Sie mich gleich klarstellen, dass einige Details vereinfacht wurden, um unnötige Komplexität zu vermeiden.

Technische Vertiefung

Als wir eine Lösung für das Problem planten, entschieden wir uns zunächst dafür, Anwendungen in Java unter Verwendung des Spring-Frameworks, des Nginx-Balancers, der Postgres-Datenbank und anderer technischer und weniger technischer Dinge zu erstellen. Da die Zeit, eine technische Lösung zu entwickeln, es uns ermöglichte, andere Ansätze zur Lösung dieses Problems in Betracht zu ziehen, fiel unser Blick auf die Apache NIFI-Technologie, die in bestimmten Kreisen in Mode ist. Ich sage gleich, dass diese Technologie es uns ermöglicht hat, diese drei Dienste zu bemerken. In diesem Artikel wird die Entwicklung eines Dateitransportdienstes und eines Datenübertragungsdienstes für den Verbraucher beschrieben. Wenn der Artikel jedoch nützlich ist, werde ich über den Dienst zum Aktualisieren von Daten in der Datenbank schreiben.

Was ist das

NIFI ist eine verteilte Architektur zum schnellen parallelen Laden und Verarbeiten von Daten, einer Vielzahl von Plugins für Quellen und Transformationen, Versionierung von Konfigurationen und vielem mehr. Ein schöner Bonus ist, dass es sehr einfach zu bedienen ist. Triviale Prozesse wie getFile, sendHttpRequest und andere können als Quadrate dargestellt werden. Jedes Quadrat stellt einen Prozess dar, dessen Zusammenwirken in der folgenden Abbildung dargestellt ist. Es wurde eine ausführlichere Dokumentation zu Prozess-Setup-Interaktionen geschrieben hier , für diejenigen, die Russisch sprechen - hier. Die Dokumentation beschreibt perfekt, wie man NIFI entpackt und ausführt und wie man Prozesse, auch Quadrate genannt, erstellt
Die Idee, einen Artikel zu schreiben, entstand nach einer langen Suche und der Strukturierung der erhaltenen Informationen in etwas Bewusstes, sowie dem Wunsch, zukünftigen Entwicklern das Leben ein wenig einfacher zu machen.

Beispiel

Es wird ein Beispiel dafür betrachtet, wie Quadrate miteinander interagieren. Das allgemeine Schema ist recht einfach: Wir erhalten eine HTTP-Anfrage (theoretisch mit einer Datei im Hauptteil der Anfrage). Um die Fähigkeiten von NIFI zu demonstrieren, startet die Anfrage in diesem Beispiel den Prozess des Empfangens einer Datei aus dem lokalen Dateispeicher ), dann senden wir eine Antwort zurück, dass die Anfrage eingegangen ist, parallel dazu erfolgt der Empfang einer Datei von FH und der anschließende Prozess des Verschiebens per FTP zu FH. Es ist klarzustellen, dass Prozesse über die sogenannte FlowFile miteinander interagieren. Dies ist die Basisentität in NIFI, die Attribute und Inhalte speichert. Inhalt sind die Daten, die durch die Streamdatei dargestellt werden. Das heißt, grob gesagt, wenn Sie eine Datei von einem Quadrat erhalten und auf ein anderes übertragen, ist der Inhalt Ihre Datei.

Wie Sie sehen können, zeigt dieses Bild den allgemeinen Prozess. HandleHttpRequest – akzeptiert Anfragen, ReplacementText – generiert einen Antworttext, HandleHttpResponse – sendet eine Antwort. FetchFile – empfängt eine Datei von einem Dateispeicher und überträgt sie an den Platz. PutSftp – legt diese Datei auf FTP an der angegebenen Adresse ab. Jetzt mehr über diesen Prozess.

In diesem Fall ist die Anfrage der Anfang von allem. Schauen wir uns die Konfigurationsparameter an.

Bis auf StandardHttpContextMap ist hier alles recht trivial – das ist eine Art Dienst, mit dem Sie Anfragen senden und empfangen können. Ausführlicher und sogar anhand von Beispielen können Sie sehen: hier

Schauen wir uns als Nächstes die Konfigurationsparameter „ReplaceText“ des Quadrats an. Es lohnt sich, auf ReplacementValue zu achten – dieser wird in Form einer Antwort an den Benutzer zurückgegeben. In den Einstellungen können Sie den Grad der Protokollierung anpassen, Sie können die Protokolle sehen {wo Sie nifi entpackt haben}/nifi-1.9.2/logs, es gibt auch Fehler-/Erfolgsparameter – basierend auf diesen Parametern können Sie den Prozess als Ganzes regulieren . Das heißt, im Falle einer erfolgreichen Textverarbeitung wird der Vorgang des Sendens einer Antwort an den Benutzer aufgerufen, in einem anderen Fall protokollieren wir lediglich den erfolglosen Vorgang.

Außer dem Status, wenn eine Antwort erfolgreich erstellt wurde, gibt es in den HandleHttpResponse-Eigenschaften nichts besonders Interessantes.

Wir haben die Anfrage und die Antwort geklärt – jetzt können wir die Datei empfangen und auf dem FTP-Server ablegen. FetchFile – empfängt eine Datei unter dem in den Einstellungen angegebenen Pfad und übergibt sie an den nächsten Prozess.

Und dann platziert das PutSftp-Quadrat die Datei im Dateispeicher. Wir können die Konfigurationsparameter unten sehen.

Es ist zu beachten, dass jedes Quadrat ein separater Prozess ist, der gestartet werden muss. Wir haben uns das einfachste Beispiel angesehen, das keine komplexe Anpassung erfordert. Als nächstes schauen wir uns den Prozess etwas komplizierter an und schreiben ein wenig auf die Rillen.

Komplexeres Beispiel

Der Datenübertragungsdienst zum Verbraucher erwies sich aufgrund der Änderung der SOAP-Nachricht als etwas komplizierter. Der allgemeine Prozess ist in der folgenden Abbildung dargestellt.

Auch hier ist die Idee nicht besonders kompliziert: Wir haben vom Verbraucher eine Anfrage erhalten, dass er Daten benötigt, eine Antwort gesendet, dass er eine Nachricht erhalten hat, den Prozess des Empfangs der Antwortdatei gestartet, sie dann mit einer bestimmten Logik bearbeitet und dann übertrug die Datei in Form einer SOAP-Nachricht an den Server an den Verbraucher.

Ich denke, dass es nicht nötig ist, die Quadrate, die wir oben gesehen haben, noch einmal zu beschreiben – gehen wir direkt zu den neuen über. Wenn Sie eine Datei bearbeiten müssen und normale Quadrate vom Typ „ReplaceText“ nicht geeignet sind, müssen Sie Ihr eigenes Skript schreiben. Dies kann mit dem ExecuteGroogyScript-Quadrat erfolgen. Die Einstellungen werden unten dargestellt.

Es gibt zwei Möglichkeiten, das Skript in dieses Quadrat zu laden. Die erste besteht darin, eine Datei mit einem Skript herunterzuladen. Die zweite Möglichkeit besteht darin, ein Skript in scriptBody einzufügen. Soweit ich weiß, unterstützt das ExecuteScript-Quadrat mehrere Sprachen – eine davon ist groovy. Ich werde Java-Entwickler enttäuschen – in solchen Quadraten kann man keine Skripte in Java schreiben. Für diejenigen, die es wirklich wollen, müssen Sie Ihr eigenes benutzerdefiniertes Quadrat erstellen und es dem NIFI-System hinzufügen. Dieser ganze Vorgang wird von einem ziemlich langen Tanz mit einem Tamburin begleitet, auf den wir in diesem Artikel nicht eingehen. Ich habe mich für die groovige Sprache entschieden. Unten finden Sie ein Testskript, das die ID in einer SOAP-Nachricht einfach inkrementell aktualisiert. Es ist wichtig zu beachten. Sie nehmen die Datei von flowFile und aktualisieren sie. Vergessen Sie nicht, dass Sie sie aktualisiert wieder dort ablegen müssen. Es ist auch erwähnenswert, dass nicht alle Bibliotheken enthalten sind. Es kann vorkommen, dass Sie noch eine der Bibliotheken importieren müssen. Ein weiterer Nachteil besteht darin, dass das Skript in diesem Bereich recht schwer zu debuggen ist. Es gibt eine Möglichkeit, eine Verbindung zur NIFI-JVM herzustellen und den Debugging-Prozess zu starten. Ich persönlich habe eine lokale Anwendung gestartet und den Empfang einer Datei aus der Sitzung simuliert. Ich habe auch lokal debuggt. Fehler, die beim Laden eines Skripts auftreten, erkennt Google recht einfach und wird von NIFI selbst in das Protokoll geschrieben.

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("Error during processing of validate.groovy", e)
    session.transfer(flowFile, REL_FAILURE)
}

Eigentlich endet hier die Individualisierung des Quadrats. Als nächstes wird die aktualisierte Datei an den Square übertragen, der für das Senden der Datei an den Server verantwortlich ist. Nachfolgend finden Sie die Einstellungen für dieses Quadrat.

Wir beschreiben die Methode, mit der eine SOAP-Nachricht übertragen wird. Wir schreiben wo. Als nächstes müssen Sie angeben, dass es sich um SOAP handelt.

Fügen Sie mehrere Eigenschaften wie Host und Aktion (soapAction) hinzu. Wir speichern und prüfen. Weitere Details zum Senden von SOAP-Anfragen finden Sie hier hier

Wir haben verschiedene Optionen für den Einsatz von NIFI-Prozessen geprüft. Wie interagieren sie und was ist ihr wirklicher Nutzen? Die betrachteten Beispiele sind Testbeispiele und unterscheiden sich geringfügig von dem, was tatsächlich im Kampf passiert. Ich hoffe, dass dieser Artikel für Entwickler ein wenig nützlich sein wird. Vielen Dank für Ihre Aufmerksamkeit. Wenn Sie Fragen haben, schreiben Sie. Ich werde versuchen zu antworten.

Source: habr.com