Hej alla!

Uppgiften består i följande - det finns en flow som visas på bilden ovan, som måste rullas ut på N servrar med Apache NiFi. Flowet är ett test - det handlar om att generera en fil och skicka den till en annan NiFi-instans. Dataöverföringen sker via NiFi Site to Site-protokollet.

NiFi Site to Site (S2S) är ett säkert, lättanpassat sätt att överföra data mellan NiFi-instanser. Hur S2S fungerar kan du läsa om i dokumentationen och det är viktigt att inte glömma att konfigurera NiFi-instansen för att tillåta S2S, se ). Ett typiskt exempel på användning:.

I de fall där det handlar om dataöverföring med S2S - en instans kallas klient, den andra server. Klienten skickar data, servern tar emot. Det finns två sätt att ställa in dataöverföring mellan dem:

1. Push. Från klientinstansen skickas data med hjälp av Remote Process Group (RPG). På serverinstansen tas data emot via Input Port
2. Pull. Servern tar emot data via RPG, klienten skickar med hjälp av Output port.

Flowet för rullning lagrar vi i Apache Registry.

Apache NiFi Registry är ett underprojekt av Apache NiFi, som representerar ett verktyg för att lagra flöden och hantera versioner. En sorts GIT. Information om installation, konfiguration och arbete med registry kan hittas i den officiella dokumentationen. Flödena för lagring samlas i en processgrupp och lagras så i registry. Vi kommer att återkomma till detta längre fram i artikeln.

I början, när N är ett litet antal, levereras och uppdateras flödet manuellt inom rimlig tid.

Men med ökande N blir problemen fler:

1. det tar längre tid att uppdatera flödet. Man måste logga in på alla servrar
2. det uppstår fel vid uppdatering av mallar. Här har vi uppdaterat, men här har vi glömt
3. mänskliga fel vid utförande av ett stort antal likadana operationer

Allt detta leder oss till att vi måste automatisera processen. Jag har testat följande lösningar på detta problem:

1. Använda MiNiFi istället för NiFi
2. NiFi CLI
3. NiPyAPI

Användning av MiNiFi

Apache MiNiFy — delprojekt Apache NiFi. MiNiFy — en kompakt agent som använder samma processorer som NiFi, vilket möjliggör skapandet av samma flöden som i NiFi. Agentens lättviktighet uppnås delvis genom att MiNiFy inte har något grafiskt gränssnitt för konfiguration av flöden. Avsaknaden av ett grafiskt gränssnitt i MiNiFy innebär att man måste lösa problemet med att leverera flöden till Minifi. Eftersom MiNiFy används aktivt inom IoT, finns det många komponenter, och processen att leverera flöden till slutliga Minifi-instans måste automatiseras. Känns det igen?

En annan delprojekt som kan hjälpa att lösa ett sådant problem är MiNiFi C2 Server. Denna produkt är avsedd att fungera som en central punkt i arkitekturen för distribution av konfigurationer. Hur man konfigurerar miljön beskrivs i denna artikel Habré, och informationen är tillräcklig för att lösa den aktuella uppgiften. MiNiFi uppdaterar automatiskt sin konfiguration i tandem med C2 server. Den enda nackdelen med detta tillvägagångssätt är att man måste skapa mallar på C2 Server, ett enkelt åtagande i registret är inte tillräckligt.

Alternativet som beskrivs i artikeln ovan är funktionellt och inte svårt att implementera, men man får inte glömma följande:

1. Det finns inte alla processorer från NiFi i MiNiFi.
2. Versionsnummerna på processorerna i MiNiFi ligger efter versionerna i NiFi.

Vid tidpunkten för skrivandet av publiceringen är den senaste versionen av NiFi — 1.9.2. Versionsnumret på processorerna i den senaste versionen av MiNiFi är 1.7.0. Processorer kan läggas till MiNiFi, men på grund av versionsskillnader mellan NiFi- och MiNiFi-processorerna kanske det inte fungerar.

NiFi CLI

Enligt beskrivningen av verktyget på den officiella webbplatsen, detta är ett verktyg för att automatisera interaktionen mellan NiFi och NiFi Registry inom området för flödesleverans eller processhantering. För att börja använda detta verktyg måste det laddas ner. härifrån.

Vi startar verktyget

.\/bin\/cli.sh
           _     ___  _
 Apache   (_)  .' ..](_)   ,
 _ .--.   __  _| |_  __    )
[ `.-. | [  |'-| |-'[  |  \/  
|  | | |  | |  | |   | | '    '
[___||__][___][___] [___]',  ,'
                           `'
          CLI v1.9.2

Skriv 'help' för att se en lista över tillgängliga kommandon, använd tab för att autocompleta.

För att ladda ner det nödvändiga flödet från registret måste vi veta bucket identifierare (bucket identifier) och flödets identifierare (flow identifier). Dessa data kan erhållas antingen via CLI eller i webgränssnittet för NiFi-registret. I webbgränssnittet ser det ut så här:

Med hjälp av CLI gör man så här:

#> registry list-buckets -u http://nifi-registry:18080

#   Name             Id                                     Description
-   --------------   ------------------------------------   -----------
1   test_bucket   709d387a-9ce9-4535-8546-3621efe38e96   (empty)

#> registry list-flows -b 709d387a-9ce9-4535-8546-3621efe38e96 -u http://nifi-registry:18080

#   Name           Id                                     Description
-   ------------   ------------------------------------   -----------
1   test_flow   d27af00a-5b47-4910-89cd-9c664cd91e85

Vi startar importen av processgruppen från registret:

#> nifi pg-import -b 709d387a-9ce9-4535-8546-3621efe38e96 -f d27af00a-5b47-4910-89cd-9c664cd91e85 -fv 1 -u http://nifi:8080

7f522a13-016e-1000-e504-d5b15587f2f3

En viktig punkt — vilken värd som helst kan anges som processen som vi tillämpar processgruppen på.

Processgruppen har lagts till med stoppade processorer, de måste startas.

#> nifi pg-start -pgid 7f522a13-016e-1000-e504-d5b15587f2f3 -u http://nifi:8080

Utmärkt, processorerna har startat. Men enligt uppgiften behöver vi att NiFi-instanserna skickar data till andra instanser. Låt oss anta att vi har valt Push-metoden för att överföra data till servern. För att organisera datatransmission måste vi aktivera överföringen på den tillagda Remote Process Group (RPG) som redan ingår i vårt flöde.

I dokumentationen för CLI och andra källor har jag inte hittat något sätt att aktivera datatransmission. Om du vet hur man gör detta — vänligen skriv det i kommentarerna.

Eftersom vi har bash och är redo att gå hela vägen — låt oss hitta en lösning! Vi kan använda NiFi API för att lösa det här problemet. Vi kommer att använda följande metod, ID tar vi från exemplen ovan (i vårt fall är det 7f522a13-016e-1000-e504-d5b15587f2f3). Beskrivning av metoderna i NiFi API. ). Ett typiskt exempel på användning:.

I body måste vi skicka JSON av följande typ:

{
    "revision": {
	    "clientId": "value",
	    "version": 0,
	    "lastModifier": "value"
	},
    "state": "value",
    "disconnectedNodeAcknowledged": true
}

Parametrarna som måste fyllas i för att det ska "fungera":
state — överföringsstatus. Tillgängligt TRANSMITTING för att aktivera datatransmission, STOPPED för att stänga av.
version — processor-version.

version kommer som standard att vara 0 vid skapande, men dessa parametrar kan erhållas med metoden.

För bash-skriptets älskare kan denna metod verka användbar, men det är inte riktigt min starka sida — bash-skript är inte min favorit. Den nästa metoden är enligt min åsikt mer intressant och bekväm.

NiPyAPI

NiPyAPI — ett bibliotek för Python för att interagera med NiFi-instanser. Dokumentationssidan innehåller nödvändig information för att arbeta med biblioteket. Snabbstart finns beskriven i projekt på github.

Vårt skript för att distribuera konfigurationen — ett program skrivet i Python. Låt oss gå vidare med kodningen.
Vi konfigurerar inställningarna för framtida arbete. Vi behöver följande parametrar:

nipyapi.config.nifi_config.host = 'http://nifi:8080/nifi-api' #väg till nifi-api-instansen där vi distribuerar processgruppen
nipyapi.config.registry_config.host = 'http://nifi-registry:18080/nifi-registry-api' #väg till nifi-registry-api-registret
nipyapi.config.registry_name = 'MyBeutifulRegistry' #namn på registret som det kommer att kallas i nifi-instansen
nipyapi.config.bucket_name = 'BucketName' #namn på bucketen från vilken vi drar flödet
nipyapi.config.flow_name = 'FlowName' #namn på flödet som vi drar in

Jag kommer nu att införa namnen på metoderna i detta bibliotek som beskrivs ). Ett typiskt exempel på användning:.

Vi ansluter registret till nifi-instansen med hjälp av

nipyapi.versioning.create_registry_client

I detta steg kan man också lägga till en kontroll för att se om registret redan har lagts till instansen, för detta kan man använda metoden

nipyapi.versioning.list_registry_clients

Vi hittar bucket för vidare sökning av flow i korgen

nipyapi.versioning.get_registry_bucket

Utifrån den hittade bucketen söker vi efter flow

nipyapi.versioning.get_flow_in_bucket

Det är viktigt att förstå om denna processgrupp redan har lagts till. Processgruppen placeras utifrån koordinater och det kan hända att en andra komponent överlappar en första. Jag har kontrollerat, så kan det vara 🙂 För att få alla tillagda processgrupper använder vi metoden

nipyapi.canvas.list_all_process_groups

och vi kan sedan söka, till exempel efter namn.

Jag kommer inte att beskriva processen för att uppdatera mallen, jag nämner bara att om processorer läggs till i den nya versionen av mallen finns det inga problem med meddelanden i köerna. Men om processorer tas bort kan problem uppstå (nifi tillåter inte att ta bort en processor om det finns en kö med meddelanden framför den). Om du är intresserad av hur jag löste detta problem - skriv till mig, så diskuterar vi detta. Kontakterna finns i slutet av artikeln. Låt oss gå vidare till steget för att lägga till processgruppen.

Vid felsökning av skriptet stötte jag på en egenskap som gör att den senaste versionen av flow inte alltid dras in, så jag rekommenderar att man först kontrollerar denna version:

nipyapi.versioning.get_latest_flow_ver

Vi deployar processgruppen:

nipyapi.versioning.deploy_flow_version

Vi startar processorerna:

nipyapi.canvas.schedule_process_group

I avsnittet om CLI stod det att dataöverföring inte automatiskt aktiveras i remote process group? När jag implementerade skriptet stötte jag också på detta problem. Vid den tidpunkten kunde jag inte starta dataöverföring via API och jag beslöt mig för att skriva till utvecklaren av NiPyAPI-biblioteket och be om råd/hjälp. Utvecklaren svarade mig, vi diskuterade problemet och han skrev att han behövde tid "för att kolla på något". Och så, efter ett par dagar kom ett brev där en funktion på Python som löser mitt problem dök upp!!! Vid den tidpunkten var versionen av NiPyAPI 0.13.3 och i den fanns så klart inget sådant. Men i version 0.14.0, som nyligen släpptes, ingick den här funktionen i biblioteket. Här är den,

nipyapi.canvas.set_remote_process_group_transmission

Så med hjälp av biblioteket NiPyAPI kopplade vi upp oss mot registry, skapade flow och till och med startade processorer och dataöverföring. Nu kan vi finslipa koden, lägga till olika kontroller, loggning och allt det där. Men det är en helt annan historia.

Av de alternativ jag har övervägt tyckte jag att det senaste var det mest funktionella. För det första handlar det fortfarande om kod i Python, vilket gör att man kan integrera hjälpprogram och dra nytta av språkets fördelar. För det andra utvecklas projektet NiPyAPI aktivt, och vid problem kan man kontakta utvecklaren. För det tredje är NiPyAPI ett mer flexibelt verktyg för att interagera med NiFi när man hanterar komplexa problem. Till exempel i att fastställa om meddelandeköerna i flow just nu är tomma och om processgruppen kan uppdateras.

Det var allt. Jag har beskrivit tre tillvägagångssätt för automatisering av leveransen av flow i NiFi, fallgropar som utvecklaren kan stöta på och gett fungerande kod för automatisering av leveransen. Om du är lika intresserad av detta ämne som jag är — hör av dig!

Källa: habr.com