Apache NIFI - En kort översikt över funktioner i praktiken

Inledning

Det hände så att jag på min nuvarande arbetsplats var tvungen att bekanta mig med denna teknik. Jag börjar med lite bakgrund. Vid nästa möte fick vårt team veta att vi behövde skapa integration med känt system. Med integration menades det att detta välkända system skulle skicka oss förfrågningar via HTTP till en specifik slutpunkt, och vi, konstigt nog, skulle skicka tillbaka svar i form av ett SOAP-meddelande. Allt verkar enkelt och trivialt. Av detta följer att du behöver...

Uppgift

Skapa 3 tjänster. Den första av dem är Databas Update Service. Den här tjänsten, när ny data kommer från ett tredjepartssystem, uppdaterar data i databasen och genererar en fil i CSV-format för att överföra den till nästa system. Slutpunkten för den andra tjänsten kallas - FTP Transport Service, som tar emot den överförda filen, validerar den och lägger den i fillagring via FTP. Den tredje tjänsten, konsumentdataöverföringstjänsten, fungerar asynkront med de två första. Den tar emot en begäran från ett externt system från tredje part att ta emot filen som diskuterats ovan, tar den färdiga svarsfilen, modifierar den (uppdaterar id, beskrivning, länkToFile-fält) och skickar svaret i form av ett SOAP-meddelande. Det vill säga, den övergripande bilden är följande: de två första tjänsterna börjar sitt arbete först när data för uppdatering har anlänt. Den tredje tjänsten fungerar konstant eftersom det finns många konsumenter av information, cirka 1000 förfrågningar om data per minut. Tjänster är ständigt tillgängliga och deras instanser finns i olika miljöer, såsom test, demo, förproduktion och prod. Nedan är ett diagram över hur dessa tjänster fungerar. Låt mig genast klargöra att vissa detaljer har förenklats för att undvika onödig komplexitet.

Teknisk fördjupning

När vi planerade en lösning på problemet bestämde vi oss först för att göra applikationer i Java med hjälp av Spring-ramverket, Nginx balancer, Postgres-databasen och andra tekniska och inte så tekniska saker. Eftersom tiden för att utveckla en teknisk lösning gjorde det möjligt för oss att överväga andra tillvägagångssätt för att lösa detta problem, föll vår blick på Apache NIFI-tekniken, som är på modet i vissa kretsar. Jag ska genast säga att den här tekniken gjorde det möjligt för oss att lägga märke till dessa 3 tjänster. Den här artikeln kommer att beskriva utvecklingen av en filtransporttjänst och en dataöverföringstjänst till konsumenten, men om artikeln är användbar kommer jag att skriva om tjänsten för att uppdatera data i databasen.

Vad är det här

NIFI är en distribuerad arkitektur för snabb parallell laddning och bearbetning av data, ett stort antal plugins för källor och transformationer, versionering av konfigurationer och mycket mer. En trevlig bonus är att den är väldigt enkel att använda. Triviala processer som getFile, sendHttpRequest och andra kan representeras som rutor. Varje ruta representerar en process, vars samverkan kan ses i figuren nedan. Mer detaljerad dokumentation om interaktioner med processinställningar har skrivits här , för dem som talar ryska - här. Dokumentationen beskriver perfekt hur man packar upp och kör NIFI, samt hur man skapar processer, även kända som kvadrater
Idén att skriva en artikel föddes efter en lång sökning och strukturering av den mottagna informationen till något medvetet, såväl som önskan att göra livet lite lättare för framtida utvecklare.

Exempel

Ett exempel på hur rutor interagerar med varandra betraktas. Det allmänna schemat är ganska enkelt: Vi tar emot en HTTP-förfrågan (I teorin, med en fil i förfrågans brödtext. För att demonstrera funktionerna hos NIFI, startar förfrågan i detta exempel processen att ta emot en fil från den lokala fillagringen ), sedan skickar vi tillbaka ett svar om att förfrågan har tagits emot, parallellt med processen att ta emot en fil från FH och sedan processen att flytta den via FTP till FH. Det är värt att förtydliga att processer interagerar med varandra genom den så kallade flowFile. Detta är basenheten i NIFI som lagrar attribut och innehåll. Innehåll är den data som representeras av stream-filen. Det vill säga, grovt sett, om du tar emot en fil från en ruta och överför den till en annan så blir innehållet din fil.

Som du kan se visar den här bilden den allmänna processen. HandleHttpRequest - accepterar förfrågningar, ReplaceText - genererar en svarskropp, HandleHttpResponse - skickar ett svar. FetchFile - tar emot en fil från en fillagring, överför den till kvadraten PutSftp - lägger denna fil på FTP, på den angivna adressen. Nu mer om denna process.

I det här fallet är begäran början på allt. Låt oss titta på dess konfigurationsparametrar.

Allt här är ganska trivialt med undantag för StandardHttpContextMap - det här är en slags tjänst som låter dig skicka och ta emot förfrågningar. Mer detaljerat och även med exempel kan du se - här

Låt oss sedan titta på ReplaceText-konfigurationsparametrarna för kvadraten. Det är värt att uppmärksamma ReplacementValue - det här är vad som kommer att returneras till användaren i form av ett svar. I inställningar kan du justera nivån på loggning, du kan se loggarna {där du packade upp nifi}/nifi-1.9.2/loggar, det finns även misslyckande/framgångsparametrar - baserat på dessa parametrar kan du reglera processen som helhet . Det vill säga, i fallet med framgångsrik textbehandling kommer processen att skicka ett svar till användaren att anropas, och i ett annat fall kommer vi helt enkelt att logga den misslyckade processen.

Det finns inget särskilt intressant i HandleHttpResponse-egenskaperna förutom statusen när ett svar har skapats.

Vi har sorterat förfrågan och svaret - låt oss gå vidare till att ta emot filen och placera den på FTP-servern. FetchFile - tar emot en fil på den sökväg som anges i inställningarna och skickar den till nästa process.

Och sedan PutSftp-rutan - placerar filen i fillagringen. Vi kan se konfigurationsparametrarna nedan.

Det är värt att uppmärksamma det faktum att varje ruta är en separat process som måste startas. Vi tittade på det enklaste exemplet som inte kräver någon komplex anpassning. Därefter kommer vi att titta på processen lite mer komplicerad, där vi kommer att skriva lite om spåren.

Mer komplext exempel

Dataöverföringstjänsten till konsumenten visade sig vara lite mer komplicerad på grund av processen att ändra SOAP-meddelandet. Den allmänna processen visas i figuren nedan.

Här är idén inte heller särskilt komplicerad: vi fick en förfrågan från konsumenten om att han behövde data, skickade ett svar att han hade fått ett meddelande, startade processen med att ta emot svarsfilen, redigerade den sedan med en viss logik och sedan överfört filen till konsumenten i form av ett SOAP-meddelande till servern.

Jag tror att det inte finns något behov av att beskriva de rutor som vi såg ovan igen - låt oss gå direkt till de nya. Om du behöver redigera någon fil och vanliga rutor av typen ReplaceText inte är lämpliga, måste du skriva ditt eget skript. Detta kan göras med hjälp av ExecuteGroogyScript-rutan. Dess inställningar presenteras nedan.

Det finns två alternativ för att ladda skriptet i denna ruta. Den första är genom att ladda ner en fil med ett skript. Den andra är genom att infoga ett skript i scriptBody. Så vitt jag vet stöder executeScript-torget flera språk - ett av dem är groovy. Jag kommer att göra java-utvecklare besvikna - du kan inte skriva skript i java i sådana rutor. För de som verkligen vill måste du skapa din egen anpassade kvadrat och lägga till den i NIFI-systemet. Hela denna operation åtföljs av en ganska lång dans med en tamburin, som vi inte kommer att behandla i den här artikeln. Jag valde det groovy språket. Nedan finns ett testskript som helt enkelt gradvis uppdaterar ID:t i ett SOAP-meddelande. Det är viktigt att notera. Du tar filen från flowFile och uppdaterar den, glöm inte att du behöver lägga tillbaka den där, uppdaterad. Det är också värt att notera att inte alla bibliotek ingår. Det kan hända att du fortfarande måste importera en av libs. En annan nackdel är att skriptet i denna ruta är ganska svårt att felsöka. Det finns ett sätt att ansluta till NIFI JVM och starta felsökningsprocessen. Personligen startade jag en lokal applikation och simulerade att jag fick en fil från sessionen. Jag gjorde också felsökning lokalt. Fel som dyker upp när ett skript laddas är ganska lätta att Google och skrivs av NIFI själv till 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)
}

Egentligen är det här anpassningen av torget slutar. Därefter överförs den uppdaterade filen till torget, som ansvarar för att skicka filen till servern. Nedan finns inställningarna för denna ruta.

Vi beskriver metoden med vilken ett SOAP-meddelande kommer att sändas. Vi skriver var. Därefter måste du ange att detta är SOAP.

Lägg till flera egenskaper som värd och action (soapAction). Vi sparar och kollar. Du kan se mer information om hur du skickar SOAP-förfrågningar här

Vi tittade på flera alternativ för att använda NIFI-processer. Hur interagerar de och vad är deras verkliga fördel? Exemplen som tas upp är testexemplen och skiljer sig något från vad som faktiskt händer i strid. Jag hoppas att den här artikeln kommer att vara lite användbar för utvecklare. Tack för din uppmärksamhet. Om du har några frågor, skriv. Jag ska försöka svara.

Källa: will.com