Tere kõigile!

Ülesanne on järgmine — on voog, nagu on näidatud ülaltoodud pildil, mille tuleb rakendada N serverisse, Apache NiFi. Voog on testvool — toimub faili genereerimine ja saatmine teise NiFi instantsi. Andmete edastamine toimub NiFi Site to Site protokolli kaudu.

NiFi Site to Site (S2S) — turvaline, lihtsalt seadistatav viis andmete edastamiseks NiFi instantside vahel. Kuidas S2S töötab, vaata dokumentatsioon ja on oluline mitte unustada seadistada NiFi instantsi, et lubada S2S, vaata siit.

Juhtudel, kui räägitakse andmete edastamisest S2S kaudu — üks instants on kliendi, teine serveri oma. Kliendi instants saadab andmeid, serveri instants vastuvõtab. Kaks viisi andmete edastamiseks nende vahel:

1. Push. Kliendi instantsist saadetakse andmeid Remote Process Group (RPG) abil. Serveri instantsis võetakse andmeid vastu Input Port kaudu.
2. Pull. Server võtab andmeid vastu RPG kaudu, klient saadab andmeid Output Port abil.

Voog, mida rakendame, on salvestatud Apache Registry's.

Apache NiFi Registry — Apache NiFi allprojekti tööriist voogude salvestamiseks ja versioonide haldamiseks. Nagu GIT. Teavet registri installimise, seadistamise ja kasutamise kohta leiate ametlikust dokumentatsioonist. Hoidmiseks mõeldud voogude grupp liidetakse protsessi grupp ja sellisel kujul salvestatakse registrisse. Tagasi sellele teemale tuleme hiljem artiklis.

Alguses, kui N on väike arv, tarnitakse ja ajakohastatakse voogu käsitsi vastuvõetava ajaga.

Aga N kasvades tekivad probleemid:

1. voo ajakohastamiseks kulub rohkem aega. Tuleb sisse logida kõikidele serveritele,
2. tekivad mallide ajakohastamise vead. Siin on uuendatud, aga seal on unustatud.
3. inimese eksimused, kui tehakse palju sarnaseid operatsioone.

Kõik see viib meid järeldusele, et protsess tuleb automatiseerida. Olen proovinud järgmisi lahendusi sellele ülesandele:

1. Kasutada MiNiFi-d NiFi asemel.
2. NiFi CLI
3. NiPyAPI

MiNiFi kasutamine

Apache MiNiFi — Apache NiFi alaprojekt. MiNiFy on kompaktne agent, mis kasutab samu protsessoreid nagu NiFi, võimaldades luua samu vooge nagu NiFi. Agendi kergekaaluline olemus tuleneb osaliselt sellest, et MiNiFy-l puudub graafiline kasutajaliides voogude konfigureerimiseks. MiNiFy graafilise kasutajaliidese puudumine tähendab, et tuleb lahendada voogude edastamise probleem minifi-sse. Kuna MiNiFy-d kasutatakse aktiivselt IOT-s, on komponente palju ja voogude edastamise protsess lõplikesse MiNiFy eksemplaridesse tuleb automatiseerida. Kas see ei tundu tuttav probleem?

Seda probleemi aitab lahendada veel üks alaprojekt — MiNiFi C2 Server. See toode on mõeldud, et olla keskses punktis konfiguratsioonide levitamise arhitektuuris. Kuidas konfigureerida keskkonda — on kirjeldatud selles artiklis Habr's, ja teave on piisav, et lahendada antud probleem. MiNiFi koos C2 serveriga värskendab oma konfiguratsiooni automaatrežiimis. Ainus selle lähenemise miinus on see, et C2 Serveris tuleb luua mallid, lihtsalt registry-sse commitimine ei ole piisav.

Ülaltoodud artiklis kirjeldatud variant on töötav ja mitte keeruline ellu viia, kuid ei tohi unustada järgmist:

1. MiNiFy-s pole kõiki protsessoreid, mis on NiFi-s
2. Protsessorite versioonid MiNiFy-s jäävad NiFi protsessorite versioonidest maha.

Selle artikli kirjutamise ajal on NiFi viimane versioon — 1.9.2. Viimane MiNiFi protsessorite versioon on — 1.7.0. Protsessoreid saab MiNiFy-sse lisada, kuid versioonide erinevuse tõttu NiFi ja MiNiFy protsessorite vahel ei pruugi see töötada.

NiFi CLI

Üksikasjalike andmete põhjal kirjeldus tööriistast ametlikul veebilehel, see on tööriist NiFi ja NiFi Registry koostöö automatiseerimiseks voogude edastamise või protsesside haldamise osas. Selle tööriista kasutamiseks tuleb see alla laadida siit.

Käivitame utiliidi

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

Type 'help' to see a list of available commands, use tab to auto-complete.

Kuna me peame vajalikku voogu registry-st alla laadima, peame teadma ämbrite identifikaatoreid (bucket identifier) ja ise voogu (flow identifier). Need andmed saab kätte kas cli või NiFi registry veebiliideses. Veebiliideses näeb see välja järgmiselt:

CLI abil tehakse järgmist:

#> 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

Käivitame process grupi importimise registry-st:

#> 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

Oluline punkt — protsessigruppide rakendamiseks võib hostina kasutada mis tahes NiFi instantsi.

Protsessi grupp on lisatud peatatud protsessoritega, need tuleb käivitada.

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

Suurepärane, protsessorid on käivitunud. Kuid ülesande tingimuste kohaselt peavad NiFi instantsid andmeid saatma teistele instantsidele. Eeldame, et andmete edastamiseks serverisse valiti Push meetod. Andmete edastamiseks tuleb lisatud Remote Process Group (RPG), mis on juba meie voos, aktiveerida edastus (Enable transmitting).

Dokumentatsioonis, CLI-s ja teistes allikates ei leidnud ma viisi edastuse aktiveerimiseks. Kui teate, kuidas seda teha, palun kirjutage kommentaarides.

Kuna meil on bash ja oleme valmis lõpuni minema — leidkem lahendus! Saame kasutada NiFi API-d selle probleemi lahendamiseks. Kasutame järgmist meetodit, ID võtame eelnevatest näidetest (meie puhul on see 7f522a13-016e-1000-e504-d5b15587f2f3). NiFi API meetodite kirjeldus. siit.

Body-s tuleb edastada JSON järgmises vormingus:

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

Parameetrid, mida tuleb täita, et "töötaks":
state — andmete edastusstatus. Saadaval on TRANSMITTING andmete edastamiseks, STOPPED edastuse väljalülitamiseks.
version — protsessori versioon.

Versioon on vaike, 0 loomisel, kuid need parameetrid saab kätte meetodi abil.

Bash skriptide huvilistele võib see meetod tunduda sobiv, kuid mulle on see keeruline — bash skriptid ei ole mu lemmikud. Järgmine meetod on huvitavam ja mugavam minu arvates.

NiPyAPI

NiPyAPI — Pythonile mõeldud teek NiFi instantsidega suhtlemiseks. Dokumentatsiooni leht sisaldab vajalikku teavet teegiga töötamiseks. Kiire alustamine on kirjeldatud projektist githubis.

Meie skript konfiguratsiooni välja laskmiseks on programm Pythonis. Liigume koodimise juurde.
Seame konfiguratsioonid edasiseks tööks. Me vajame järgmisi parameetreid:

nipyapi.config.nifi_config.host = 'http://nifi:8080/nifi-api' # tee nifi-api instantsi juurde, kus protsessi gruppi arendame
nipyapi.config.registry_config.host = 'http://nifi-registry:18080/nifi-registry-api' # tee nifi-registry-api registrisse
nipyapi.config.registry_name = 'MyBeutifulRegistry' # registri nimi, kuidas see nifi instantsis nimi on
nipyapi.config.bucket_name = 'BucketName' # bucketi nimi, kust toome flow
nipyapi.config.flow_name = 'FlowName' # flow nimi, which we are pulling

Edasi hakkan sisestama selle teegi meetodite nimesid, mis on kirjeldatud siit.

Ühendame registri nifi instantsiga kasutades

nipyapi.versioning.create_registry_client

Selle sammu juures saab veel lisada kontrolli, et registri instantsile on juba lisatud, selleks võib kasutada meetodit

nipyapi.versioning.list_registry_clients

Leidke bucket edasise vooge otsimiseks prügikastis

nipyapi.versioning.get_registry_bucket

Leitud bucket'i põhjal otsime vooge

nipyapi.versioning.get_flow_in_bucket

Edasi on oluline mõista, kas see protsessigrupp on juba lisatud. Protsessigrupp paigutatakse koordinaatide põhjal ja võib juhtuda, et ühe komponendi peale asetub teine. Ma olen seda kontrollinud, selline olukord on tõepoolest võimalik 🙂. Kõikide lisatud protsessigruppide saamiseks kasutame meetodit

nipyapi.canvas.list_all_process_groups

ja seejärel võime otsida näiteks nime järgi.

Ma ei hakka šablooni uuendamise protsessi kirjeldama, vaid ütlen vaid, et kui uues šablooni versioonis lisatakse protsessorid, siis ei ole tekkinud sõnumite järjekordades probleeme. Kuid kui protsessorid eemaldatakse, võivad probleemid tekkida (nifi ei luba protsessorit eemaldada, kui selle ees on sõnumite järjekord). Kui teid huvitab, kuidas ma selle probleemi lahendasin, kirjutage mulle, palun, arutame seda küsimust. Kontaktid on artikli lõpus. Jätkame protsessigruppi lisamise sammuga.

Skripti tõrkeotsingul olen kokku puutunud eripäraga, et mitte alati ei tõugata viimast versiooni vooge, seega soovitan kõigepealt selle versiooni täpsustada:

nipyapi.versioning.get_latest_flow_ver

Deploy'ime protsessigruppi:

nipyapi.versioning.deploy_flow_version

Käivitame protsessorid:

nipyapi.canvas.schedule_process_group

CLI jaotises oli kirjutatud, et kaugprotsessigrupp ei lülita andmete edastamist automaatselt sisse? Skripti rakendamisel puutusin ma selle probleemiga samuti kokku. Sel hetkel ei õnnestunud mul API abil andmete edastamist käivitada ja otsustasin kirjutada NiPyAPI arendajale ning küsida nõu/abi. Arendaja vastas mulle, me arutasime probleemi ja ta kirjutas, et tal on vaja aega "mõned asjad kontrollida". Ja nii, paar päeva hiljem tuli kiri, milles oli Pythonis funktsioon, mis lahendab minu käivitamise probleemi!!! Sel hetkel oli NiPyAPI versioon 0.13.3 ja selles ei olnud loomulikult midagi sellist. Kuid versioonis 0.14.0, mis on ilmunud üsna hiljuti, on see funktsioon juba raamatukogus olemas. Tere tulemast,

nipyapi.canvas.set_remote_process_group_transmission

Nii, kasutades NiPyAPI teeki, ühendasime registry, tõmbasime töövoo ja isegi käivitasime protsessorid ja andmete edastamise. Edasi saab koodi lihvida, lisada erinevaid kontrolle, logimist ja nii edasi. Kuid see on juba hoopis teine lugu.

Minu läbi vaadatud automatiseerimise variantidest tundus viimane kõige toimivam. Esiteks, see on ikkagi Python keeles kood, kuhu saab integreerida abikoodi ja kasutada kõikide programmeerimiskeele eeliseid. Teiseks, NiPyAPI projekt areneb aktiivselt ja probleemide korral saab arendajale kirjutada. Kolmandaks, NiPyAPI on siiski paindlikum tööriist NiFi-ga keerukate ülesannete lahendamiseks. Näiteks, et määrata, kas sõnumijärjekorrad on praegu tühjad ja kas process group'i saab uuendada.

Sellega on kõik. Kirjeldasin kolme lähenemist töövoo automatiseerimisele NiFis, takistusi, millega arendaja võib kokku puutuda, ja esitasin töötava koodi automatiseerimiseks. Kui teid huvitab see teema, nagu mind — kirjutage!

Allikas: habr.com