Apache NIFI - Een kort overzicht van kansen in de praktijk

Introductie

Het gebeurde zo dat ik op mijn huidige werkplek kennis moest maken met deze technologie. Ik zal beginnen met een beetje achtergrond. Tijdens de volgende bijeenkomst kreeg ons team te horen dat we een integratie met bekend systeem. Met integratie werd bedoeld dat dit bekende systeem ons via HTTP verzoeken naar een specifiek eindpunt zou sturen, en wij, vreemd genoeg, antwoorden terug zouden sturen in de vorm van een SOAP-bericht. Alles lijkt eenvoudig en triviaal. Hieruit volgt dat je nodig hebt...

Taak

Maak 3 diensten. De eerste daarvan is de Database Update Service. Deze service werkt, wanneer er nieuwe gegevens binnenkomen van een systeem van derden, de gegevens in de database bij en genereert een bestand in CSV-formaat om dit over te dragen naar het volgende systeem. Het eindpunt van de tweede service wordt de FTP Transport Service genoemd, die het overgedragen bestand ontvangt, valideert en via FTP in de bestandsopslag plaatst. De derde service, de consumentengegevensoverdrachtservice, werkt asynchroon met de eerste twee. Het ontvangt een verzoek van een extern systeem van derden om het hierboven besproken bestand te ontvangen, neemt het kant-en-klare responsbestand, wijzigt het (werkt de ID-, beschrijving- en linkToFile-velden bij) en verzendt het antwoord in de vorm van een SOAP-bericht. Dat wil zeggen, het algemene beeld is als volgt: de eerste twee services beginnen pas met hun werk als de gegevens voor het bijwerken zijn gearriveerd. De derde dienst werkt constant omdat er veel informatieverbruikers zijn, ongeveer 1000 verzoeken om data per minuut. Services zijn voortdurend beschikbaar en hun instances bevinden zich in verschillende omgevingen, zoals test-, demo-, pre-productie- en productieomgeving. Hieronder ziet u een diagram van hoe deze services werken. Laat ik meteen verduidelijken dat sommige details zijn vereenvoudigd om onnodige complexiteit te voorkomen.

Technische verdieping

Bij het plannen van een oplossing voor het probleem hebben we eerst besloten om applicaties in Java te maken met behulp van het Spring-framework, de Nginx-balancer, de Postgres-database en andere technische en niet zo technische dingen. Omdat de tijd om een ​​technische oplossing te ontwikkelen ons in staat stelde andere benaderingen te overwegen om dit probleem op te lossen, viel onze blik op de Apache NIFI-technologie, die in bepaalde kringen in de mode is. Ik zal meteen zeggen dat deze technologie ons in staat heeft gesteld deze drie diensten op te merken. Dit artikel beschrijft de ontwikkeling van een service voor bestandstransport en een service voor gegevensoverdracht naar de consument, maar als het artikel nuttig is, zal ik schrijven over de service voor het bijwerken van gegevens in de database.

Wat is dit

NIFI is een gedistribueerde architectuur voor het snel parallel laden en verwerken van gegevens, een groot aantal plug-ins voor bronnen en transformaties, versiebeheer van configuraties en nog veel meer. Een leuke bonus is dat het heel gemakkelijk te gebruiken is. Triviale processen zoals getFile, sendHttpRequest en andere kunnen worden weergegeven als vierkanten. Elk vierkant vertegenwoordigt een proces, waarvan de interactie in onderstaande figuur te zien is. Er is meer gedetailleerde documentatie geschreven over de interacties met procesinstellingen hier , voor degenen die Russisch spreken - hier. De documentatie beschrijft perfect hoe je NIFI uitpakt en uitvoert, en hoe je processen creëert, ook wel vierkanten genoemd
Het idee om een ​​artikel te schrijven ontstond na lang zoeken en structureren van de ontvangen informatie tot iets bewusts, evenals de wens om het leven van toekomstige ontwikkelaars een beetje gemakkelijker te maken.

Voorbeeld

Er wordt een voorbeeld overwogen van hoe vierkanten met elkaar omgaan. Het algemene schema is vrij eenvoudig: we ontvangen een HTTP-verzoek (in theorie met een bestand in de hoofdtekst van het verzoek. Om de mogelijkheden van NIFI te demonstreren, start het verzoek in dit voorbeeld het proces van het ontvangen van een bestand van de lokale bestandsopslag ), dan sturen we een antwoord terug dat het verzoek is ontvangen, parallel aan het proces van het ontvangen van een bestand van FH en vervolgens het proces van het verplaatsen ervan via FTP naar FH. Het is de moeite waard om te verduidelijken dat processen met elkaar interacteren via het zogenaamde flowFile. Dit is de basisentiteit in NIFI waarin attributen en inhoud worden opgeslagen. Inhoud zijn de gegevens die worden weergegeven door het streambestand. Dat wil zeggen dat als u een bestand van het ene plein ontvangt en het naar het andere overbrengt, de inhoud uw bestand zal zijn.

Zoals u kunt zien, toont deze afbeelding het algemene proces. HandleHttpRequest - accepteert verzoeken, ReplaceText - genereert een antwoordtekst, HandleHttpResponse - verzendt een antwoord. FetchFile - ontvangt een bestand uit een bestandsopslag, brengt het over naar de vierkante PutSftp - plaatst dit bestand op FTP, op het opgegeven adres. Nu meer over dit proces.

In dit geval is verzoek het begin van alles. Laten we eens kijken naar de configuratieparameters.

Alles hier is vrij triviaal, met uitzondering van StandardHttpContextMap - dit is een soort service waarmee je verzoeken kunt verzenden en ontvangen. Meer gedetailleerd en zelfs met voorbeelden kunt u zien: hier

Laten we vervolgens eens kijken naar de ReplaceText-configuratieparameters van het vierkant. Het is de moeite waard om aandacht te besteden aan ReplacementValue - dit is wat in de vorm van een antwoord aan de gebruiker zal worden geretourneerd. In de instellingen kun je het logniveau aanpassen, je kunt de logs zien {waar je nifi hebt uitgepakt}/nifi-1.9.2/logs, er zijn ook fout-/succesparameters - op basis van deze parameters kun je het proces als geheel regelen . Dat wil zeggen dat in het geval van een succesvolle tekstverwerking het proces van het verzenden van een antwoord naar de gebruiker wordt aangeroepen, en in een ander geval zullen we eenvoudigweg het niet-succesvolle proces registreren.

Er is niets bijzonders interessants in de HandleHttpResponse-eigenschappen, behalve de status wanneer een antwoord met succes is aangemaakt.

We hebben het verzoek en het antwoord uitgezocht - laten we verder gaan met het ontvangen van het bestand en het plaatsen ervan op de FTP-server. FetchFile - ontvangt een bestand op het pad dat is opgegeven in de instellingen en geeft het door aan het volgende proces.

En dan plaatst het PutSftp-vierkant het bestand in de bestandsopslag. We kunnen de configuratieparameters hieronder zien.

Het is de moeite waard om aandacht te besteden aan het feit dat elk vierkant een afzonderlijk proces is dat moet worden gelanceerd. We hebben gekeken naar het eenvoudigste voorbeeld dat geen ingewikkelde aanpassingen vereist. Vervolgens zullen we het proces wat ingewikkelder bekijken, waarbij we een beetje over de groeven zullen schrijven.

Complexer voorbeeld

De gegevensoverdrachtservice naar de consument bleek iets ingewikkelder vanwege het proces van het aanpassen van het SOAP-bericht. Het algemene proces wordt weergegeven in de onderstaande afbeelding.

Ook hier is het idee niet bijzonder ingewikkeld: we ontvingen een verzoek van de consument dat hij gegevens nodig had, stuurden een antwoord dat hij een bericht had ontvangen, startten het proces van het ontvangen van het responsbestand, bewerkten het vervolgens met een bepaalde logica, en vervolgens bracht het bestand in de vorm van een SOAP-bericht naar de server over naar de consument.

Ik denk dat het niet nodig is om de vierkanten die we hierboven zagen opnieuw te beschrijven - laten we meteen naar de nieuwe gaan. Als u een bestand moet bewerken en gewone vierkanten van het type ReplaceText niet geschikt zijn, zult u uw eigen script moeten schrijven. Dit kan gedaan worden met behulp van het ExecuteGroogyScript-vierkant. De instellingen worden hieronder weergegeven.

Er zijn twee opties om het script in dit vierkant te laden. De eerste is door een bestand met een script te downloaden. De tweede is door een script in scriptBody in te voegen. Voor zover ik weet ondersteunt het executeScript-vierkant verschillende talen - een daarvan is groovy. Ik zal Java-ontwikkelaars teleurstellen - je kunt in zulke vierkanten geen scripts in Java schrijven. Voor degenen die dat echt willen, moet je je eigen vierkant op maat maken en dit toevoegen aan het NIFI-systeem. Deze hele operatie gaat gepaard met een behoorlijk lange dans met een tamboerijn, waar we in dit artikel niet op ingaan. Ik heb voor de groovy taal gekozen. Hieronder staat een testscript dat de ID in een SOAP-bericht eenvoudig stapsgewijs bijwerkt. Het is belangrijk om op te merken. Je haalt het bestand uit flowFile en werkt het bij, vergeet niet dat je het daar weer moet plaatsen, bijgewerkt. Het is ook vermeldenswaard dat niet alle bibliotheken zijn opgenomen. Het kan voorkomen dat u één van de libs nog moet importeren. Een ander nadeel is dat het script in dit vierkant vrij moeilijk te debuggen is. Er is een manier om verbinding te maken met de NIFI JVM en het foutopsporingsproces te starten. Persoonlijk heb ik een lokale applicatie gelanceerd en gesimuleerd dat ik een bestand van de sessie ontving. Ik heb ook lokaal debuggen gedaan. Fouten die verschijnen bij het laden van een script zijn voor Google vrij eenvoudig en worden door NIFI zelf naar 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 vierkant. Vervolgens wordt het bijgewerkte bestand overgebracht naar het plein, dat verantwoordelijk is voor het verzenden van het bestand naar de server. Hieronder staan ​​de instellingen voor dit vierkant.

We beschrijven de methode waarmee een SOAP-bericht wordt verzonden. Wij schrijven waar. Vervolgens dient u aan te geven dat dit SOAP is.

Voeg verschillende eigenschappen toe, zoals host en action (soapAction). Wij bewaren en controleren. U kunt meer details bekijken over het verzenden van SOAP-verzoeken hier

We hebben verschillende mogelijkheden bekeken om NIFI-processen in te zetten. Hoe werken ze met elkaar samen en wat is hun werkelijke voordeel? De beschouwde voorbeelden zijn testvoorbeelden en verschillen enigszins van wat er feitelijk gebeurt in de strijd. Ik hoop dat dit artikel een beetje nuttig zal zijn voor ontwikkelaars. Bedankt voor uw aandacht. Als je vragen hebt, schrijf dan. Ik zal proberen te antwoorden.

Bron: www.habr.com