Pozdrav svima!

Zadatak je sljedeći — postoji flow prikazan na slici iznad koji treba implementirati na N servera s Apache NiFi. Flow je testni — generira se datoteka i šalje u drugi instancu NiFi. Prijenos podataka se odvija putem protokola NiFi Site to Site.

NiFi Site to Site (S2S) — siguran, lako podesiv način prijenosa podataka između instanci NiFi. Kako S2S funkcionira pogledajte u dokumentaciju i važno je ne zaboraviti podesiti instancu NiFi da omogući S2S, pogledajte ovdje.

U slučajevima kada se radi o prijenosu podataka putem S2S — jedna instanca se naziva klijentska, a druga serverska. Klijentska šalje podatke, serverska — prima. Dva načina za postavljanje prijenosa podataka između njih su:

1. Push. S klijentske instance podaci se šalju putem Remote Process Group (RPG). Na serverskoj instanci podaci se primaju putem Input Port.
2. Pull. Server prima podatke putem RPG-a, klijent šalje putem Output port.

Flow za implementaciju čuvamo u Apache Registry.

Apache NiFi Registry — podprojekt Apache NiFi koji predstavlja alat za pohranu flow i upravljanje verzijama. Nešto poput GIT-a. Informacije o instalaciji, konfiguraciji i radu s registry možete pronaći u službene dokumentacije. Flow za pohranu okuplja se u process group i na taj se način pohranjuje u registry. O tome ćemo se dodatno pozabaviti u članku.

Na početku, kada je N mala brojka, flow se isporučuje i ažurira ručno u prihvatljivom vremenu.

No s rastom N, problema postaje više:

1. a ažuriranje flow-a traje više vremena. Morate ući na sve servere
2. javljuju se greške u ažuriranju predložaka. Ovdje je ažurirano, a ovdje je zaboravljeno
3. ljudske greške pri obavljanju velikog broja istovjetnih operacija

Sve to nas navodi na to da je potrebno automatizirati proces. Pokušao sam sljedeće načine rješavanja ovog zadatka:

1. Koristiti MiNiFi umjesto NiFi
2. NiFi CLI
3. NiPyAPI

Korištenje MiNiFi

Apache MiNiFy — podprojekt Apache NiFi. MiNiFy — kompaktni agent koji koristi iste procesore kao NiFi, omogućujući stvaranje istih flow-ova kao u NiFi-ju. Lakša priroda agenta postignuta je dijelom i zbog toga što MiNiFy nema grafičko sučelje za konfiguraciju flow-a. Nedostatak grafičkog sučelja u MiNiFy-ju znači da je potrebno riješiti problem isporuke flow-a do MiNiFy-ja. Budući da se MiNiFy aktivno koristi u IOT-u, postoji mnogo komponenti i proces isporuke flow-a do krajnjih instanci MiNiFy-ja treba automatizirati. Poznata zadaća, zar ne?

Rješenje takve zadaće pomoći će još jedan podprojekt — MiNiFi C2 Server. Ovaj proizvod osmišljen je kao središnja točka u arhitekturi raspodjele konfiguracija. Kako konfigurirati okruženje — opisano je u ovom članku na Habr-u i informacija je dovoljna za rješavanje postavljene zadaće. MiNiFi u suradnji s C2 serverom automatski ažurira svoju konfiguraciju. Jedini nedostatak takvog pristupa je što je potrebno stvarati predloške na C2 Serveru, jednostavno komitiranje u registry nije dovoljno.

Varijanta opisana u gornjem članku radi i nije komplicirana za implementaciju, ali treba imati na umu sljedeće:

1. U MiNiFy-ju nema svih procesora iz NiFi-ja.
2. Verzije procesora u MiNiFy-ju zaostaju za verzijama procesora u NiFi-ju.

U trenutku pisanja publikacije, posljednja verzija NiFi-ja je 1.9.2. Verzija procesora najnovije verzije MiNiFi-ja je 1.7.0. Procesore je moguće dodati u MiNiFi, ali zbog razlika u verzijama između procesora NiFi i MiNiFi to možda neće raditi.

NiFi CLI

Sudeći po opisu alata na službenoj stranici, to je alat za automatizaciju interakcije između NiFi-ja i NiFi Registry-ja u području isporuke flow-a ili upravljanja procesima. Za početak rada, ovaj alat je potrebno preuzeti odavde.

Pokrećemo alat

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

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

Da bismo učitali potrebni flow iz registry-ja, moramo znati identifikatore košarice (bucket identifier) i samog flow-a (flow identifier). Ove podatke možemo dobiti ili putem CLI-a ili u web-sučeju NiFi registry-ja. U web sučelju izgleda ovako:

Pomoću CLI se radi ovako:

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

Pokrećemo uvoz process group-a iz registry-ja:

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

Važna stavka — kao host na koji primjenjujemo process group može biti bilo koja instanca nifi.

Process group je dodan s zaustavljenim procesorima, potrebno ih je pokrenuti.

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

Odlično, procesori su se pokrenuli. Međutim, prema uvjetima zadatka, trebamo da instancije NiFi šalju podatke na druge instance. Pretpostavimo da smo za prijenos podataka na server odabrali metodu Push. Da bismo organizirali prijenos podataka, trebamo na dodanom Remote Process Group (RPG), koji je već uključen u naš flow, uključiti prijenos podataka (Enable transmitting).

U dokumentaciji, CLI i drugim izvorima nisam našao način za uključivanje prijenosa podataka. Ako znate kako to učiniti — molim vas, napišite u komentarima.

Ako već imamo bash i spremni smo ići do kraja — pronađimo rješenje! Možemo iskoristiti NiFi API za rješavanje ovog problema. Iskoristit ćemo sljedeću metodu, ID uzimamo iz prethodnih primjera (u našem slučaju to je 7f522a13-016e-1000-e504-d5b15587f2f3). Opis metoda NiFi API. ovdje.

U body trebate poslati JSON sljedećeg oblika:

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

Parametri koje je potrebno ispuniti kako bi “začelo”:
stanje — status prijenosa podataka. Dostupno TRANSMITTING za uključivanje prijenosa podataka, STOPPED za isključivanje.
version — verzija procesora.

verzija će po defaultu biti 0 prilikom stvaranja, ali te parametre možete dobiti koristeći metodu.

Za ljubitelje bash skripti ova metoda može se činiti korisnom, ali meni je teška — bash skripte nisu moj najomiljeniji. Sljedeći način je zanimljiviji i praktičniji po mom mišljenju.

NiPyAPI

NiPyAPI — knjižnica za Python jezik za interakciju s instancama NiFi. Stranica s dokumentacijom sadrži potrebne informacije za rad s knjižnicom. Quick start opisan je u projektu na github-u.

Naš skript za implementaciju konfiguracije — program na Python jeziku. Prelazimo na kodiranje.
Postavljamo konfiguracije za daljnji rad. Trebat će nam sljedeći parametri:

nipyapi.config.nifi_config.host = 'http://nifi:8080/nifi-api' #putanja do nifi-api instance na kojoj implementiramo process group
nipyapi.config.registry_config.host = 'http://nifi-registry:18080/nifi-registry-api' #putanja do nifi-registry-api registry
nipyapi.config.registry_name = 'MyBeutifulRegistry' #naziv registry, kako će se zvati u instanci nifi
nipyapi.config.bucket_name = 'BucketName' #naziv bucket-a, iz kojeg povlačimo flow
nipyapi.config.flow_name = 'FlowName' #naziv flow-a koji povlačimo

Dalje ću umetnuti nazive metoda ove biblioteke koji su opisani ovdje.

Povezujemo registry s instancom nifi pomoću

nipyapi.versioning.create_registry_client

Na ovom koraku još možemo dodati provjeru da li je registry već dodan instanci, za to možemo iskoristiti metodu

nipyapi.versioning.list_registry_clients

Pronalazimo bucket za daljnje pretraživanje flowa u košarici

nipyapi.versioning.get_registry_bucket

Prema pronađenom bucketu tražimo flow

nipyapi.versioning.get_flow_in_bucket

Dalje je važno razumjeti je li ovaj process group već dodat. Process group se postavlja na koordinate i može se dogoditi situacija kada se jedan komponenta preklapa s drugom. Provjeravao sam, to se može dogoditi 🙂 Da bismo dobili sve dodane process group, koristimo metodu

nipyapi.canvas.list_all_process_groups

i zatim možemo tražiti, na primjer po imenu.

Neću opisivati proces ažuriranja predloška, reći ću samo da ako se u novoj verziji predloška dodaju procesori, nije problem s prisutnošću poruka u redovima. No ako se procesori uklone, može doći do problema (nifi ne dopušta brisanje procesora ako se ispred njega nakupila reda poruka). Ako vas zanima kako sam riješio taj problem — molim vas, javite mi se, razgovarat ćemo o tome. Kontakti su na kraju članka. Prelazimo na korak dodavanja process group.

Prilikom debagiranja skripte naletio sam na posebnost da se ne prikazuje uvijek najnovija verzija flowa, stoga preporučujem prvo pojasniti ovu verziju:

nipyapi.versioning.get_latest_flow_ver

Deployamo process group:

nipyapi.versioning.deploy_flow_version

Pokrećemo procesore:

nipyapi.canvas.schedule_process_group

U odjeljku o CLI-u pisalo je da se u remote process group automatski ne uključuje prijenos podataka? Pri realizaciji skripte i ja sam se suočio s tim problemom. Tada nisam mogao pokrenuti prijenos podataka pomoću API-ja i odlučio sam se obratiti razvojnim programerima biblioteke NiPyAPI za savjet/pomoć. Razvijač mi je odgovorio, razgovarali smo o problemu i rekao je da mu treba vremena "da provjeri nešto". I evo, nakon nekoliko dana dolazi pismo s funkcijom na Pythonu koja rješava moj problem pokretanja!!! U to vrijeme verzija NiPyAPI bila je 0.13.3 i u njoj, naravno, nije bilo ničega takvog. No u verziji 0.14.0, koja je izašla nedavno, ta funkcija je već uključena u biblioteku. Upoznajte,

nipyapi.canvas.set_remote_process_group_transmission

Dakle, pomoću biblioteke NiPyAPI povezali smo registry, primenili flow i čak pokrenuli procesore i prenos podataka. Dalje možemo srediti kod, dodati razne provere, logovanje i sve to. No, to je već potpuno druga priča.

Od opcija za automatizaciju koje sam razmatrao, poslednja mi se pokazala kao najfunkcionalnija. Prvo, to je svakako kod u Pythonu, u koji možemo integrisati pomoćni programski kod i iskoristiti sve prednosti programskog jezika. Drugo, projekat NiPyAPI se aktivno razvija i u slučaju problema možemo kontaktirati developera. Treće, NiPyAPI je ipak fleksibilniji alat za interakciju sa NiFijem u rešavanju složenih zadataka. Na primer, u određivanju da li su redovi poruka trenutno prazni u flowu i možemo li ažurirati grupu procesa.

To je sve. Opišao sam 3 pristupa ka automatizaciji isporuke flowa u NiFi, potencijalne zamke sa kojima se developer može susresti i dao radni kod za automatizaciju isporuke. Ako vas ova tema zanima kao i mene — javite se!

Izvor: habr.com