Apache NIFI — Een kort overzicht van de mogelijkheden in de praktijk.

Inleiding

Het is zo gebeurd dat ik op mijn huidige werkplek met deze technologie in aanraking ben gekomen. Laat me beginnen met een korte achtergrond. bekende systeem. Met integratie werd bedoeld dat dit bekende systeem ons via HTTP verzoeken zou sturen naar een bepaalde endpoint, en wij, zoals het vreemd lijkt, antwoorden zouden terugsturen in de vorm van een SOAP-bericht. Het lijkt allemaal eenvoudig en triviaal. Hieruit volgt dat we…

Task

Drie services moeten creëren. De eerste is de Database Update Service. Deze service, bij ontvangst van nieuwe gegevens uit een extern systeem, actualiseert de gegevens in de database en genereert een bestand in CSV-indeling om het naar het volgende systeem te verzenden. De endpoint van de tweede service — de Transport Service via FTP — wordt aangeroepen, die het overgedragen bestand ontvangt, valideert en in de bestandsopslag plaatst via FTP. De derde service — de Data Delivery Service — werkt asynchroon met de eerste twee. Hij ontvangt een verzoek van een extern systeem om het bestand waar we het eerder over hadden te ontvangen, neemt het gereedstaande responsbestand, wijzigt het (werkt de velden id, description, linkToFile bij) en stuurt een antwoord in de vorm van een SOAP-bericht. Kortom, het geheel ziet er als volgt uit: de eerste twee services beginnen hun werk alleen wanneer er gegevens voor bijwerking binnenkomen. De derde service werkt continu, omdat er veel informatieontvangers zijn, ongeveer 1000 verzoeken om gegevens per minuut. De services zijn constant beschikbaar en hun instanties bevinden zich op verschillende omgevingen, zoals test, demo, pre-productie en productie. Hieronder is een schema van de werking van deze services weergegeven. Ik zal onmiddellijk verduidelijken dat sommige details zijn vereenvoudigd om onnodige complexiteit te vermijden.

Technische verdieping

Bij het plannen van de oplossing voor het probleem, werd besloten om applicaties in Java te maken met het Spring-framework, een load balancer Nginx, een Postgres-database en andere technische en minder technische zaken. Aangezien de tijd voor het uitwerken van de technische oplossing ruimte bood om andere benaderingen te overwegen, viel de blik op de populaire technologie Apache NIFI die in bepaalde kringen gebruikt wordt. Ik zeg meteen dat deze technologie ons in staat heeft gesteld deze 3 diensten te identificeren. In dit artikel beschrijf ik de ontwikkeling van de bestandstransportdienst en de dienst voor gegevensoverdracht naar de gebruiker, maar als het artikel aanslaat, schrijf ik ook over de dienst voor het bijwerken van gegevens in de database.

Wat is dit?

NIFI is een gedistribueerde architectuur voor snelle, parallelle gegevensinvoer en -verwerking, met een groot aantal plug-ins voor bronnen en transformaties, versiebeheer van configuraties en nog veel meer. Een prettige bonus is dat het erg eenvoudig te gebruiken is. Triviale processen zoals getFile, sendHttpRequest en anderen kunnen worden weergegeven in de vorm van vierkanten. Elk vierkant vertegenwoordigt een bepaald proces, waarvan de interactie kan worden gezien in de afbeelding hieronder. Gedetailleerde documentatie over de interactie en het instellen van processen is geschreven hier , voor degenen die het in het Russisch willen — hier. In de documentatie staat uitstekend beschreven hoe je NIFI uitpakt en start, en hoe je processen kunt maken, die ook vierkanten zijn.
Het idee om dit artikel te schrijven kwam voort uit langdurige zoektochten en het structureren van de verkregen informatie tot iets begrijpelijk, evenals de wens om het leven van toekomstige ontwikkelaars een beetje gemakkelijker te maken.

Voorbeeld

Hier is een voorbeeld van hoe de vierkanten met elkaar interageren. Het algemene schema is vrij eenvoudig: We ontvangen een HTTP-verzoek (theoretisch met een bestand in de lichaam van het verzoek. Voor de demonstratie van de mogelijkheden van NIFI, start in dit voorbeeld het verzoek het proces om een bestand uit de lokale opslag te verkrijgen), vervolgens wordt er een antwoord verzonden dat het verzoek is ontvangen, tegelijkertijd wordt het proces voor het verkrijgen van het bestand uit de opslag gestart en daarna het proces voor het verplaatsen ervan via FTP naar de opslag. Het is belangrijk te vermelden dat de processen met elkaar communiceren via de zogenaamde flowFile. Dit is de basisentiteit in NIFI, die attributen en inhoud opslaat. De inhoud zijn de gegevens die in het stroombestand worden gepresenteerd. Dit betekent grofweg dat als je een bestand van het ene vierkant ontvangt en het naar een ander doorgeeft, de inhoud jouw bestand zal zijn.

Zoals je kunt opmerken - op deze afbeelding is het algemene proces afgebeeld. HandleHttpRequest - ontvangt verzoeken, ReplaceText - genereert de inhoud van het antwoord, HandleHttpResponse - geeft het antwoord terug. FetchFile - haalt een bestand uit de bestandopslag en geeft het door aan het vierkant PutSftp - plaatst dit bestand op de FTP op het opgegeven adres. Laten we nu dit proces nader bekijken.

In dit geval is de request de basis. Laten we kijken naar de configuratieparameters.

Hier is alles vrij triviaal, met uitzondering van de StandartHttpContextMap - dit is een soort service die het mogelijk maakt om verzoeken te verzenden en te ontvangen. Meer gedetailleerd, zelfs met voorbeelden, kan je hier kijken - hier

Laten we verder kijken naar de configuratieparameters van het ReplaceText-vierkant. Let vooral op ReplacementValue - dat is wat aan de gebruiker wordt teruggegeven als antwoord. In de instellingen kan je het niveau van logging aanpassen, de logs zijn te vinden {waar nifi uitgepakt is}\/nifi-1.9.2\/logs, daar zijn ook parameters voor failure/success - op basis van deze parameters kan het proces in het algemeen worden geregeld. Dat wil zeggen, in het geval van een succesvolle verwerking van de tekst - zal het proces voor het verzenden van een antwoord aan de gebruiker worden aangeroepen, en in andere gevallen loggen we gewoon het onsuccesvolle proces.

In de eigenschappen van HandleHttpResponse is er verder niets interessants te vermelden, behalve de status bij het succesvol aanmaken van een antwoord.

We hebben het over het verzoek en het antwoord gehad - laten we verder gaan met het verkrijgen van het bestand en het plaatsen ervan op de FTP-server. FetchFile - haalt een bestand op de opgegeven locatie in de instellingen en geeft het door aan het volgende proces.

En vervolgens plaatst de PutSftp-vierkant het bestand in de bestandsopslag. De configuratieparameters kunnen we hieronder zien.

Het is belangrijk op te merken dat elk vierkant een afzonderlijk proces is dat moet worden gestart. We hebben het eenvoudigste voorbeeld bekeken dat geen complexe aanpassingen vereist. Laten we nu een iets complexer proces bekijken, waarbij we een beetje met Groovy gaan schrijven.

Complexer voorbeeld

De datatransferdienst naar de consument is iets complexer geworden door het proces van het modificeren van SOAP-berichten. Het algemene proces is hieronder weergegeven.

Hier is het idee ook niet bijzonder ingewikkeld: we kregen een aanvraag van de consument voor gegevens, stuurden een antwoord dat we het bericht hadden ontvangen, startten het proces om het antwoordbestand te verkrijgen, bewerkten het vervolgens met bepaalde logica en zonden het bestand naar de consument in de vorm van een SOAP-bericht naar de server.

Ik denk niet dat het nodig is om opnieuw de vierkanten te beschrijven die we hierboven hebben gezien — laten we direct doorgaan naar nieuwe. Als je enig bestand wilt bewerken en gewone vierkanten zoals ReplaceText niet geschikt zijn, moet je je eigen script schrijven. Dit kan worden gedaan met behulp van het ExecuteGroovyScript-vierkant. De instellingen daarvan worden hieronder weergegeven.

Er zijn twee manieren om een script in dit blok te laden. De eerste is door een bestand met het script te uploaden. De tweede is door het script in de scriptBody in te voegen. Voor zover ik weet, ondersteunt het blok executeScript verschillende programmeertalen - een daarvan is Groovy. Java-ontwikkelaars zullen teleurgesteld zijn - je kunt geen scripts in zulke blokken in Java schrijven. Voor degenen die echt willen, moet je je eigen aangepaste blok maken en deze in het NIFI-systeem integreren. Deze operatie gaat gepaard met vrij langdurige dansen met een trommel, waar we in dit artikel niet op ingaan. Ik heb de taal Groovy gekozen. Hieronder staat een testscript dat gewoon incrementeel de id in het SOAP-bericht bijwerkt. Het is belangrijk op te merken dat je een bestand uit flowFile haalt, het bijwerkt, en vergeet niet om het, wanneer het is bijgewerkt, weer terug te plaatsen. Ook is het vermeldenswaard dat niet alle bibliotheken zijn aangesloten. Het kan zo zijn dat je toch een van de libs moet importeren. Een andere nadeel is dat het script in dit blok behoorlijk moeilijk te debuggen is. Er is een manier om verbinding te maken met de JVM van NIFI en het debugproces te starten. Persoonlijk heb ik mijn lokale applicatie uitgevoerd en gemodelleerd hoe ik een bestand uit de sessie ontvang. De debugging heb ik ook lokaal uitgevoerd. Fouten die optreden bij het uploaden van het script zijn vrij gemakkelijk op te zoeken via Google en worden door NIFI zelf in de log geschreven.

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)
}

Eigenlijk eindigt hier de aanpassing van het blok. Vervolgens wordt het bijgewerkte bestand doorgegeven aan het blok dat verantwoordelijk is voor het verzenden van het bestand naar de server. Hieronder zijn de instellingen van dit blok weergegeven.

We beschrijven de methode waarmee het SOAP-bericht zal worden verzonden. We schrijven waarheen. Vervolgens moeten we aangeven dat het specifiek SOAP is.

We voegen een aantal eigenschappen toe, zoals host en actie(soapAction). We slaan op, controleren. Meer details over het verzenden van SOAP-verzoeken zijn te vinden. here

We hebben enkele manieren bekeken waarop NIFI-processen worden gebruikt. Hoe ze met elkaar communiceren en wat de echte voordelen zijn. De besproken voorbeelden zijn testvoorbeelden en wijken enigszins af van wat er in de praktijk gebeurt. Ik hoop dat dit artikel nuttig zal zijn voor ontwikkelaars. Bedankt voor uw aandacht. Als u vragen heeft, neem gerust contact op. Ik zal mijn best doen om te antwoorden.

Bron: habr.com