Apache NIFI - En kort oversikt over muligheter i praksis

Innledning

Det hendte at jeg på min nåværende arbeidsplass måtte bli kjent med denne teknologien. Jeg starter med litt bakgrunn. På neste møte fikk teamet vårt beskjed om at vi måtte skape integrasjon med kjent system. Med integrasjon var det ment at dette velkjente systemet skulle sende oss forespørsler via HTTP til et spesifikt endepunkt, og vi, merkelig nok, ville sende tilbake svar i form av en SOAP-melding. Alt virker enkelt og trivielt. Av dette følger det at du trenger...

Oppgave

Opprett 3 tjenester. Den første av dem er Database Update Service. Denne tjenesten, når nye data kommer fra et tredjepartssystem, oppdaterer dataene i databasen og genererer en fil i CSV-format for å overføre den til neste system. Endepunktet til den andre tjenesten kalles - FTP Transport Service, som mottar den overførte filen, validerer den og legger den i fillagring via FTP. Den tredje tjenesten, forbrukerdataoverføringstjenesten, fungerer asynkront med de to første. Den mottar en forespørsel fra et eksternt tredjepartssystem om å motta filen diskutert ovenfor, tar den klare svarfilen, endrer den (oppdaterer id, beskrivelse, linkToFile-felt) og sender svaret i form av en SOAP-melding. Det vil si at det overordnede bildet er som følger: de to første tjenestene begynner sitt arbeid først når dataene for oppdatering har kommet. Den tredje tjenesten fungerer konstant fordi det er mange forbrukere av informasjon, ca 1000 forespørsler om data per minutt. Tjenestene er tilgjengelig hele tiden og deres instanser er lokalisert i forskjellige miljøer, som test, demo, pre-produksjon og prod. Nedenfor er et diagram over hvordan disse tjenestene fungerer. La meg presisere med en gang at noen detaljer er forenklet for å unngå unødvendig kompleksitet.

Teknisk utdyping

Da vi planla en løsning på problemet, bestemte vi oss først for å lage applikasjoner i Java ved å bruke Spring-rammeverket, Nginx balancer, Postgres-database og andre tekniske og ikke så tekniske ting. Siden tiden for å utvikle en teknisk løsning tillot oss å vurdere andre tilnærminger for å løse dette problemet, falt blikket vårt på Apache NIFI-teknologien, som er mote i visse kretser. Jeg vil si med en gang at denne teknologien tillot oss å legge merke til disse 3 tjenestene. Denne artikkelen vil beskrive utviklingen av en filtransporttjeneste og en dataoverføringstjeneste til forbrukeren, men hvis artikkelen er nyttig, vil jeg skrive om tjenesten for oppdatering av data i databasen.

Hva er dette

NIFI er en distribuert arkitektur for rask parallell lasting og prosessering av data, et stort antall plugins for kilder og transformasjoner, versjonering av konfigurasjoner og mye mer. En fin bonus er at den er veldig enkel å bruke. Trivielle prosesser som getFile, sendHttpRequest og andre kan representeres som firkanter. Hver rute representerer en prosess, hvis interaksjon kan sees i figuren nedenfor. Mer detaljert dokumentasjon om interaksjoner med prosessoppsett er skrevet her , for de som snakker russisk - her. Dokumentasjonen beskriver perfekt hvordan du pakker ut og kjører NIFI, samt hvordan du lager prosesser, også kjent som firkanter
Ideen om å skrive en artikkel ble født etter et langt søk og strukturering av informasjonen mottatt til noe bevisst, samt ønsket om å gjøre livet litt lettere for fremtidige utviklere.

Eksempel

Et eksempel på hvordan firkanter samhandler med hverandre vurderes. Det generelle opplegget er ganske enkelt: Vi mottar en HTTP-forespørsel (I teorien, med en fil i forespørselens brødtekst. For å demonstrere mulighetene til NIFI, starter forespørselen i dette eksemplet prosessen med å motta en fil fra den lokale fillagringen ), så sender vi tilbake et svar om at forespørselen er mottatt, parallelt prosessen med å motta en fil fra FH og deretter prosessen med å flytte den via FTP til FH. Det er verdt å presisere at prosesser samhandler med hverandre gjennom den såkalte flowFile. Dette er basisenheten i NIFI som lagrer attributter og innhold. Innhold er dataene som er representert av strømfilen. Det vil si, grovt sett, hvis du mottar en fil fra en rute og overfører den til en annen, vil innholdet være din fil.

Som du kan se, viser dette bildet den generelle prosessen. HandleHttpRequest - godtar forespørsler, ReplaceText - genererer en svartekst, HandleHttpResponse - sender et svar. FetchFile - mottar en fil fra en fillagring, overfører den til firkanten PutSftp - legger denne filen på FTP, på den angitte adressen. Nå mer om denne prosessen.

I dette tilfellet er forespørselen begynnelsen på alt. La oss se på konfigurasjonsparametrene.

Alt her er ganske trivielt med unntak av StandardHttpContextMap - dette er en slags tjeneste som lar deg sende og motta forespørsler. Mer detaljert og til og med med eksempler kan du se - her

Deretter, la oss se på ReplaceText-konfigurasjonsparametrene til kvadratet. Det er verdt å ta hensyn til ReplacementValue - dette er det som vil bli returnert til brukeren i form av et svar. I innstillinger kan du justere nivået på logging, du kan se loggene {hvor du pakket ut nifi}/nifi-1.9.2/logger, det er også feil/suksessparametere - basert på disse parameterne kan du regulere prosessen som helhet . Det vil si at i tilfelle vellykket tekstbehandling, vil prosessen med å sende et svar til brukeren bli kalt, og i et annet tilfelle vil vi ganske enkelt logge den mislykkede prosessen.

Det er ingenting spesielt interessant i HandleHttpResponse-egenskapene bortsett fra statusen når et svar er vellykket opprettet.

Vi har sortert ut forespørselen og svaret - la oss gå videre til å motta filen og plassere den på FTP-serveren. FetchFile - mottar en fil på banen spesifisert i innstillingene og sender den til neste prosess.

Og deretter PutSftp-firkanten - plasserer filen i fillageret. Vi kan se konfigurasjonsparametrene nedenfor.

Det er verdt å ta hensyn til det faktum at hver firkant er en egen prosess som må startes. Vi så på det enkleste eksemplet som ikke krever noen komplisert tilpasning. Deretter skal vi se på prosessen litt mer komplisert, hvor vi vil skrive litt om sporene.

Mer komplekst eksempel

Dataoverføringstjenesten til forbrukeren viste seg å være litt mer komplisert på grunn av prosessen med å endre SOAP-meldingen. Den generelle prosessen er vist i figuren nedenfor.

Her er ideen heller ikke spesielt komplisert: vi fikk en forespørsel fra forbrukeren om at han trengte data, sendte et svar om at han hadde mottatt en melding, startet prosessen med å motta svarfilen, så redigerte den med en viss logikk, og deretter overførte filen til forbrukeren i form av en SOAP-melding til serveren.

Jeg tror det ikke er nødvendig å beskrive igjen de rutene vi så ovenfor - la oss gå rett til de nye. Hvis du trenger å redigere en fil og vanlige ReplaceText-firkanter ikke er egnet, må du skrive ditt eget skript. Dette kan gjøres ved å bruke ExecuteGroogyScript-firkanten. Innstillingene er presentert nedenfor.

Det er to alternativer for å laste skriptet inn i denne firkanten. Den første er ved å laste ned en fil med et skript. Den andre er ved å sette inn et skript i scriptBody. Så vidt jeg vet, støtter executeScript-firkanten flere språk - ett av dem er groovy. Jeg vil skuffe java-utviklere - du kan ikke skrive skript i java i slike firkanter. For de som virkelig vil, må du lage din egen tilpassede firkant og legge den til NIFI-systemet. Hele denne operasjonen er ledsaget av en ganske lang dans med en tamburin, som vi ikke vil behandle i denne artikkelen. Jeg valgte det groovy språket. Nedenfor er et testskript som ganske enkelt gradvis oppdaterer ID-en i en SOAP-melding. Det er viktig å merke seg. Du tar filen fra flowFile og oppdaterer den, ikke glem at du må legge den tilbake der, oppdatert. Det er også verdt å merke seg at ikke alle biblioteker er inkludert. Det kan hende at du fortsatt må importere en av libs. En annen ulempe er at skriptet i denne firkanten er ganske vanskelig å feilsøke. Det er en måte å koble til NIFI JVM og starte feilsøkingsprosessen. Personlig lanserte jeg en lokal applikasjon og simulerte mottak av en fil fra økten. Jeg gjorde også feilsøking lokalt. Feil som dukker opp ved lasting av et skript er ganske enkelt å Google og skrives av NIFI selv til loggen.

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

Egentlig er det her tilpasningen av firkanten slutter. Deretter overføres den oppdaterte filen til torget, som er ansvarlig for å sende filen til serveren. Nedenfor er innstillingene for denne firkanten.

Vi beskriver metoden som en SOAP-melding vil bli overført på. Vi skriver hvor. Deretter må du indikere at dette er SOAP.

Legg til flere egenskaper som host og action (soapAction). Vi sparer og sjekker. Du kan se flere detaljer om hvordan du sender SOAP-forespørsler her

Vi så på flere alternativer for bruk av NIFI-prosesser. Hvordan samhandler de og hva er deres virkelige fordel? Eksemplene som vurderes er testeksempler og er litt forskjellige fra det som faktisk skjer i kamp. Jeg håper denne artikkelen vil være litt nyttig for utviklere. Takk for din oppmerksomhet. Hvis du har spørsmål, skriv. Jeg skal prøve å svare.

Kilde: www.habr.com