Tere kõigile!

Ülesanne on järgmine — on voog, nagu on näidatud ülaltoodud pildil, mille tuleb jagada N-serveritesse Apache NiFi. Voog on katsev vorm — toimub faili genereerimine ja edastamine teise NiFi instantsi. Andmete edastamine toimub NiFi Site to Site protokolli kaudu.

NiFi Site to Site (S2S) — turvaline, kergesti seadistatav viis andmete edastamiseks NiFi instantside vahel. Kuidas S2S töötab, vaata dokumentatsioonis ja on oluline mitte unustada seadistada NiFi instantsi S2S lubamiseks, vaata siin.

Asetades S2S andmeedastust, nimetatakse üks instants kliendiks ja teine serveriks. Kliendi instants saadab andmeid, serveri instants aktsepteerib. Andmete edastamiseks nende vahel on kaks võimalust:

1. Push. Kliendi instantsis edastatakse andmed kaudu Remote Process Group (RPG). Serveri instantsis andmed aktsepteeritakse Input Port kaudu
2. Tõmba. Server aktsepteerib andmeid RPG kaudu, klient saadab Output port kaudu.

Voogude jagamine toimub Apache Registrys.

Apache NiFi Registry — Apache NiFi alaprojekt, mis esindab voogude salvestamise ja versioonide haldamise tööriista. Nagu GIT. Teavet registri installimise, seadistamise ja kasutamise kohta leiate ametlikus dokumentatsioonis. Flow salvestamine koondatakse process group'i ja hoitakse sellisel kujul registris. Edasi liigume artiklis selle juurde veel.

Alguses, kui N on väike number, toimetatakse flow käsitsi ja ajakohastatakse mõistliku ajaga.

Aga N suurenedes kerkib esile rohkem probleeme:

1. flow ajakohastamiseks kulub rohkem aega. Peab sisse logima kõikidele serveritele
2. tekivad mallide ajakohastamise vead. Üks on siin uuendatud, aga siia unustati
3. inimese vead suure hulga sarnaste toimingute tegemisel

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

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

MiNiFi kasutamine

Apache MiNiFi — Apache NiFi alamhanke. MiNiFy on kompaktne agent, mis kasutab samu protsessoreid nagu NiFi ja võimaldab luua samu vooge, nagu NiFi. Agendi kerge kaal saavutatakse osaliselt selle tõttu, et MiNiFy-l ei ole voogude konfigureerimiseks graafilist kasutajaliidest. MiNiFy graafilise liidese puudumine tähendab, et voogude edastamise probleem tuleb lahendada minifi kohta. Kuna MiNiFy-d kasutatakse aktiivselt IoT-s, on komponente palju ja voogude edastamise automatiseerimine lõpp-instantsidesse minifi on vajalik. Kas see ei tundu tuttav ülesanne?

Seda ülesannet aitab lahendada veel üks alamprojekt — MiNiFi C2 Server. See toode on mõeldud olema keskne punkt konfiguratsioonide laadimise arhitektuuris. Kuidas keskkonda konfigureerida, on kirjas selles artiklis Habr's ja seal on piisavalt teavet esitatud ülesande lahendamiseks. MiNiFi koos C2 serveriga värskendab oma konfiguratsiooni automaatselt. Ainsaks selle lähenemise puuduseks on see, et C2 Serveris tuleb luua malle, lihtne commit registrisse ei ole piisav.

Ülaltoodud artiklis kirjeldatud variant on töökindel ning mitte keeruline rakendada, kuid tuleb meeles pidada järgmist:

1. Minifi-s ei ole kõiki NiFi protsessoreid.
2. Minifi protsessorite versioonid jäävad NiFi protsessorite versioonidest maha.

Käesoleva avalduse kirjutamise ajal on NiFi viimane versioon 1.9.2. MiNiFi viimase versiooni protsessorite versioon on 1.7.0. Protsessoreid saab MiNiFi-sse lisada, kuid NiFi ja MiNiFi protsessorite versioonide erinevuse tõttu ei pruugi see toimida.

NiFi CLI

Vastavalt kirjeldus tööriista ametlikul veebisaidil, see on tööriist NiFi ja NiFi Registry vahelise suhtluse automatiseerimiseks, et toimetada töötlejat või hallata protsesse. 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.

Selleks, et me saaksime vajaliku töötleja registry-st laadida, peame teadma ämber identifikaatoreid (bucket identifier) ja töötleja identifikaatoreid (flow identifier). Need andmed saab kas CLI kaudu või NiFi registri veebiliideses. Veebiliideses näeb see välja selline:

CLI abil tehakse nii:

#> 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 protsessigruppi 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 — hostina, millele me rakendame process group, võib olla ükskõik milline nifi instants.

Process group 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äivitatud. Kuid ülesande tingimuste kohaselt peavad NiFi instantsid saatma andmeid teistele instantsidele. Oletame, et andmete edastamiseks valiti Push meetod. Andmete edastamise korraldamiseks tuleb lisatud Remote Process Group (RPG), mis on juba meie voos, aktiveerida andmete edastamine (Enable transmitting).

Dokumentatsioonis CLI-s ja teistes allikates ei leidnud ma viisi andmete edastamise 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, et selle probleemi lahendada. Kasutame järgmist meetodit, ID võetakse eespoolt toodud näidetest (meie puhul on see 7f522a13-016e-1000-e504-d5b15587f2f3). NiFi API meetodite kirjeldus. siin.

Body-s tuleb edastada JSON järgmisel kujul:

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

Parameetrid, mida tuleb täita, et "töötaks":
state — andmeedastuse olek. Saadaolevad väärtused: TRANSMITTING andmete edastamiseks, STOPPED andmete edastamise peatamiseks.
version — protsessori versioon

versioon on vaikimisi 0 loomisel, kuid neid parameetreid saab hankida meetodi abil.

Bashi skriptide entusiastidele võib see meetod tunduda kasulik, kuid mulle isiklikult see ei meeldi — bash skriptid ei ole mu lemmikud. Järgmine meetod on huvitavam ja mugavam.

NiPyAPI

NiPyAPI — Python'i raamatukogu NiFi instantside haldamiseks. Dokumentatsiooni leht sisaldab vajalikku teavet raamatukoguga töötamiseks. Kiire alustamine on kirjas projektis githubis.

Meie konfiguratsiooni levitamise skript on Python'i programm. Liigume kodeerimise juurde.
Seadistame konfiguratsioonid edasiseks tööks. Meile on vajalikud järgmised parameetrid:

nipyapi.config.nifi_config.host = 'http://nifi:8080/nifi-api' # tee nifi-api instantsi, millel arendame protsessigruppi
nipyapi.config.registry_config.host = 'http://nifi-registry:18080/nifi-registry-api' # tee nifi-registry-api registrile
nipyapi.config.registry_name = 'MyBeautifulRegistry' # registreerimise nimi, kuidas see instantsis nifi nimedatakse
nipyapi.config.bucket_name = 'BucketName' # kasti nimi, millelt tõmbame voogu
nipyapi.config.flow_name = 'FlowName' # voolu nimi, mille tõmbame

Edasi hakkan lisama selle teegi meetodite nimetusi, mis on kirjeldatud siin.

Ühendame registry nifi instantsiga kasutades

nipyapi.versioning.create_registry_client

Selles etapis võiks veel lisada kontrolli, et registry on juba instantsiga lisatud, selleks saab kasutada meetodit

nipyapi.versioning.list_registry_clients

Otsime bucketit edasise flow otsimise jaoks prügikastist

nipyapi.versioning.get_registry_bucket

Leitud bucket'i alusel otsime flow'd

nipyapi.versioning.get_flow_in_bucket

Edasi on oluline mõista, ega see process group ei ole juba lisatud. Process group paigutatakse koordinaatide järgi ja võib juhtuda olukord, kus ühe komponendi peale kattub teine. Olen kontrollinud, see on võimalik 🙂 Kõik lisatud process group'id saame kätte kasutades meetodit

nipyapi.canvas.list_all_process_groups

ja edasi saame näiteks otsida nime järgi.

Ma ei hakka template uuendamise protsessi kirjeldama; ütlen vaid, et kui uues versioonis lisatakse protsessoreid, ei teki järjekordade sõnumite probleemide. Kui aga protsessoreid eemaldatakse, võivad tekkida probleemid (nifi ei luba protsessorit eemaldada, kui selle ees on järjekordade sõnumid). Kui teid huvitab, kuidas ma selle probleemi lahendasin, kirjutage mulle palun, arutame seda. Kontaktid on artikli lõpus. Liigume edasi process groupi lisamise sammu juurde.

Skripti silumise käigus kohtasin omadust, et viimast flow versiooni ei pruugita alati kätte saada, seetõttu soovitan esmalt selle versiooni täpsustada:

nipyapi.versioning.get_latest_flow_ver

Deploime process groupi:

nipyapi.versioning.deploy_flow_version

Käivitame protsessorid:

nipyapi.canvas.schedule_process_group

CLI jaotises oli öeldud, et remote process group ei hõlma automaatselt andmete edastamist? Koos skripti elluviimisega sattusin ka mina sellele probleemile. Sel hetkel ei õnnestunud mul andmete edastamist API abil alustada, seetõttu otsustasin küsida abi NiPyAPI raamatukogu arendajalt. Arendaja vastas mulle, me arutasime probleemi ja ta ütles, et vajab aega „mõne asja kontrollimiseks”. Ja nüüd, paar päeva hiljem, sain kirja, milles oli funktsioon Pythonis, mis lahendab minu käivitamise probleemi!!! Sel ajal oli NiPyAPI versioon 0.13.3 ja seal ei olnud muidugi midagi sellist. Kuid versioonis 0.14.0, mis ilmus täiesti hiljuti, see funktsioon juba sisaldus raamatukogus. Tere tulemast,

nipyapi.canvas.set_remote_process_group_transmission

Nii et NiPyAPI raamatukogu abil ühendasime registri, lisasime vooskeemi ja isegi käivitasime protsessorid ning andmete edastamise. Edasi saab koodi lihvida, lisada erinevaid kontrolle, logimist ja kõike muud. Aga see on juba hoopis teine lugu.

Minu arvates on viimase automaatimise võimaluse puhul tegemist kõige toimivama lahendusega. Esiteks, see on siiski Pythonis kirjutatud kood, kuhu on võimalik integreerida abiprogrammide kood ning kasutada kõiki programmeerimiskeele eeliseid. Teiseks, NiPyAPI projekt areneb aktiivselt ja probleemide korral on võimalik arendajale kirjutada. Kolmandaks, NiPyAPI on paindlikum vahend NiFi-ga keeruliste ülesannete lahendamiseks, näiteks selle määramine, kas sõnumijärjekorrad on praegu tühjad ja kas process group'i saab uuendada.

Sellega on kõik. Kirjeldasin kolme lähenemist flow edastamise automatiseerimisele NiFi-s, võimalikud takistused, millega arendaja võib kokku puutuda, ja esitasin töötava koodi automatiseerimise jaoks. Kui teid, nagu mind, huvitab see teema — kirjutage!

Allikas: habr.com