ProHoster > Blog > Administracija > Automatizacija isporuke flow u Apache NiFi

Automatizacija isporuke flow u Apache NiFi

Zdravo svima!

Automatizacija isporuke flow u Apache NiFi

Naloga je naslednja — obstaja flow, predstavljen na zgornji sliki, ki ga je treba razdeliti na N strežnikov z Apache NiFi. Flow je testni — poteka generiranje datoteke in pošiljanje v drugo instanco NiFi. Prenos podatkov poteka s pomočjo protokola NiFi Site to Site.

NiFi Site to Site (S2S) je varen, enostavno nastavljiv način prenosa podatkov med instancami NiFi. Kako deluje S2S si oglejte v dokumenti in pomembno je, da ne pozabite nastaviti instanco NiFi, da omogoči S2S, si oglejte tukaj.

V primerih, ko gre za prenos podatkov s pomočjo S2S — ena instanca se imenuje odjemalska, druga strežniška. Odjemalska pošilja podatke, strežniška jih sprejema. Dva načina za nastavitev prenosa podatkov med njima:

  1. Potisni. Z odjemalske instance se podatki pošiljajo s pomočjo Remote Process Group (RPG). Na strežniški instanci se podatki sprejemajo s pomočjo Input Port
  2. Vlečni. Strežnik sprejema podatke s pomočjo RPG, odjemalec pošilja s pomočjo Output port.


Flow za razporeditev hranimo v Apache Registry.

Apache NiFi Registry je podprojekt Apache NiFi, ki predstavlja orodje za shranjevanje flow in upravljanje različic. Nekakšen GIT. Informacije o namestitvi, nastavitvi in delu z registry lahko najdete v uradni dokumentaciji. Flow za shranjevanje se združi v process group in se v tej obliki hrani v registry. K temu se bomo v članku še vrnili.

Na začetku, ko je N majhno število, se flow dostavi in posodobi ročno v sprejemljivem času.

Toda z naraščanjem N se težave povečujejo:

  1. za posodobitev flow potrebuje več časa. Treba je obiskati vse strežnike
  2. pojavijo se napake pri posodabljanju predlog. Tu je bilo posodobljeno, a tu je bilo pozabljeno
  3. napake človeka pri izvajanju velikega števila enakih operacij

Vse to nas pripelje do tega, da je treba avtomatizirati postopek. Poskusil sem naslednje načine reševanja te naloge:

  1. Uporabiti MiNiFi namesto NiFi
  2. NiFi CLI
  3. NiPyAPI

Uporaba MiNiFi

Apache MiNiFi — podprojekt Apache NiFi. MiNiFy je kompaktný agent, ktorý využíva rovnaké procesory ako NiFi, umožňujúci vytvárať rovnaké toky ako v NiFi. Ľahkosť agenta sa dosahuje aj tým, že MiNiFy nemá grafické rozhranie na konfiguráciu toku. Absencia grafického rozhrania v MiNiFy znamená, že je potrebné vyriešiť problém dodávky toku do minifi. Keďže MiNiFy sa aktívne používa v IoT, je veľa komponentov a proces dodávky toku k finálnym inštanciám minifi je potrebné automatizovať. Známa úloha, však?

Riešenie tejto úlohy pomôže ďalší podprojekt — MiNiFi C2 Server. Tento produkt je určený na to, aby bol centrálnym bodom v architektúre nasadzovania konfigurácií. Ako nastaviť prostredie - je popísané v this article na Habr a informácie sú dostatočné na riešenie stanovenej úlohy. MiNiFi v spojení s C2 serverom automaticky aktualizuje svoju konfiguráciu. Jediný nevýhodou tohto prístupu je, že je potrebné vytvárať šablóny na C2 Serveri, jednoduchý commit do registry nestačí.

Variant popísaný v článku vyššie je funkčný a nie je zložitý na realizáciu, ale nesmieme zabudnúť na nasledujúce:

  1. V minifi nie sú všetky procesory z nifi
  2. Verzie procesorov v Minifi zaostávajú za verziami procesorov v NiFi.

V čase písania publikácie bola posledná verzia NiFi — 1.9.2. Verzia procesorov poslednej verzie MiNiFi — 1.7.0. Procesory je možné pridávať do MiNiFi, ale kvôli rozdielom vo verziách medzi procesormi NiFi a MiNiFi to nemusí fungovať.

NiFi CLI

Súdiac podľa popisu nástroja na oficiálnej stránke, je to nástroj na automatizáciu interakcie NiFi a NiFi Registry v oblasti dodávky toku alebo správy procesov. Na začiatok práce je potrebné tento nástroj stiahnuť od tod.

Spúšťame utilitu

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

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

Aby sme načítali potrebný tok z registry, musíme poznať identifikátory koša (bucket identifier) a samotného toku (flow identifier). Tieto údaje môžeme získať buď cez cli, alebo v webovom rozhraní NiFi registry. Vo webovom rozhraní to vyzerá takto:

Automatizacija isporuke flow u Apache NiFi

Pomocou CLI to robíme takto:

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

Spúšťame import procesnej skupiny z registry:

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

Pomembna točka - kot gostitelj, na katerega bomo nalagali procesno skupino, je lahko določen kateri koli primer nifi.

Procesna skupina je bila dodana z ustavljenimi procesorji, te je treba zagnati.

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

Odlično, procesorji so se zagnali. Vendar pa morajo instančni NiFi v skladu z nalogo podatke pošiljati na druge instančne naprave. Predpostavimo, da smo za prenos podatkov na strežnik izbrali način Push. Da bi omogočili prenos podatkov, moramo na dodani Remote Process Group (RPG), ki je že vključen v naš tok, omogočiti prenos podatkov (Enable transmitting).

Automatizacija isporuke flow u Apache NiFi

V dokumentaciji v CLI in drugih virih nisem našel načina za vključitev prenosa podatkov. Če veste, kako to storiti, prosim, zapišite v komentarjih.

Ker imamo bash in smo pripravljeni iti do konca - poiščimo rešitev! Lahko uporabimo NiFi API za rešitev tega problema. Uporabili bomo naslednjo metodo, ID vzamemo iz zgornjih primerov (v našem primeru je to 7f522a13-016e-1000-e504-d5b15587f2f3). Opis metod NiFi API. tukaj.

Automatizacija isporuke flow u Apache NiFi
V body je treba poslati JSON naslednjega videza:

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

Parametri, ki jih je treba izpolniti, da bo "delovalo":
state – status prenosa podatkov. Na voljo je TRANSMITTING za vključitev prenosa podatkov, STOPPED za izklop.
version – različica procesorja.

različica bo privzeto 0 ob ustvarjanju, vendar lahko te parametre dobimo z uporabo metode.

Automatizacija isporuke flow u Apache NiFi

Za ljubitelje bash skript se ta metoda morda zdi koristna, vendar je zame težka - bash skripti niso moja najljubša stvar. Naslednji način je bolj zanimiv in priročen po mojem mnenju.

NiPyAPI

NiPyAPI - knjižnica za jezik Python za interakcijo z instancami NiFi. Stran z dokumentacijo vsebuje potrebne informacije za delo s knjižnico. Hiter začetek je opisan v. projektu на github.

Naša skripta za uvajanje konfiguracije - program v jeziku Python. Gremo k kodiranju.
Nastavimo konfiguracije za nadaljnje delo. Potrebovali bomo naslednje parametre:

nipyapi.config.nifi_config.host = 'http://nifi:8080/nifi-api' #pot do nifi-api instance, na kateri razvijamo procesno skupino
nipyapi.config.registry_config.host = 'http://nifi-registry:18080/nifi-registry-api' #pot do nifi-registry-api registry
nipyapi.config.registry_name = 'MyBeutifulRegistry' #ime registry, kot se bo imenovalo v instanci nifi
nipyapi.config.bucket_name = 'BucketName' #ime bucketa, iz katerega potegnemo flow
nipyapi.config.flow_name = 'FlowName' #ime flow, ki ga potegnemo

Nadalje bom vstavljao imena metod te knjižnice, ki so opisani tukaj.

Povezujemo registry z instanco nifi s pomočjo

nipyapi.versioning.create_registry_client

V tem koraku lahko dodamo še preverjanje, ali je registry že dodan instanci, za to lahko uporabimo metodo

nipyapi.versioning.list_registry_clients

Najdemo bucket za nadaljnje iskanje flow v košu

nipyapi.versioning.get_registry_bucket

Po najdenem bucketu iščemo flow

nipyapi.versioning.get_flow_in_bucket

Nadalje je pomembno razumeti, ali je ta process group že dodan. Process group se postavi po koordinatah in lahko se zgodi, da se na en komponent nalaga drugi. Preverjal sem, to se lahko zgodi 🙂 Da pridobimo vse dodane process groups, uporabimo metodo

nipyapi.canvas.list_all_process_groups

in nato lahko iščemo, na primer po imenu.

Ne bom opisoval postopka posodobitve predloge, rekel bom le, da če v novi različici predloge procesorji dodani, s sporočili v čakalnih vrstah ni težav. Če pa se procesorji odstranijo, se lahko pojavijo težave (nifi ne dovoli odstranjevati procesorja, če se je pred njim nabrala čakalna vrsta sporočil). Če vas zanima, kako sem to težavo rešil — mi, prosim, pišite, da bi se ogovorili o tem. Kontakti so na koncu članka. Poglejmo korak dodajanja process group.

Med odpravljanjem napak skripte sem naletel na posebnost, da ne vedno prinese zadnje različice flow, zato priporočam, da najprej potrdite to različico:

nipyapi.versioning.get_latest_flow_ver

Deplojujemo process group:

nipyapi.versioning.deploy_flow_version

Zaganjamo procesorje:

nipyapi.canvas.schedule_process_group

V razdelku o CLI je bilo napisano, da v remote process group ni avtomatskega prenosa podatkov? Pri izvajanju skripte sem se tudi sam srečal s to težavo. Takrat pri izvedbi prenosa podatkov preko API mi ni uspelo in sem se odločil, da napišem razvijalcu knjižnice NiPyAPI in ga vprašam za nasvet/pomoč. Razvijalec mi je odgovoril, o težavi smo se pogovorili in napisal je, da potrebuje čas, da 'preveri nekaj'. In tako, po nekaj dneh pride e-pošta, v kateri je funkcija v Pythonu, ki rešuje mojo težavo z zagonoma!!! Takrat je bila različica NiPyAPI 0.13.3 in v njej, seveda, ni bilo nič takega. V različico 0.14.0, ki je izšla pred kratkim, pa je ta funkcija že bila vključena v knjižnico. Spoznajte,

nipyapi.canvas.set_remote_process_group_transmission

Torej, s knjižnico NiPyAPI smo povezali registry, namestili flow in celo zagnali procesorje ter prenos podatkov. Nato lahko optimiziramo kodo, dodamo različne preverbe, beleženje in vse to. A to je že povsem druga zgodba.

Od razmotrenih možnosti avtomatizacije se mi je zadnja zdela najbolj funkcionalna. Prvič, gre za kodo v pythonu, v katero lahko vgradimo pomožno programsko kodo in izkoristimo vse prednosti programskega jezika. Drugič, projekt NiPyAPI se aktivno razvija in v primeru težav lahko pišete razvijalcu. Tretjič, NiPyAPI je vseeno bolj fleksibilno orodje za interakcijo z NiFi pri reševanju kompleksnih nalog. Na primer, pri določanju ali so sedaj čakalne vrste sporočil prazne v flow in ali lahko posodobimo process group.

S tem je vse. Opisal sem 3 pristope k avtomatizaciji dostave flow v NiFi, podvodne kamenčke, s katerimi se lahko sreča razvijalec, in priložil delujočo kodo za avtomatizacijo dostave. Če vas ta tema zanima tako kot mene — pišite!

Izvor: habr.com

Kupite zanesljivo gostovanje za spletne strani z zaščito pred DDoS, VPS VDS strežniki 🔥 Kupite zanesljivo gostovanje za spletne strani z zaščito pred DDoS, VPS VDS strežniki | ProHoster