ProHoster > Blog > Administrácia > Automatizácia dodávky flow v Apache NiFi

Automatizácia dodávky flow v Apache NiFi

Ahojte všetci!

Automatizácia dodávky flow v Apache NiFi

Úloha spočíva v nasledovnom — existuje flow, ktorý je zobrazený na obrázku vyššie, ktorý treba nasadiť na N serveroch s Apache NiFi. Flow je testovací — prebieha generácia súboru a jeho odosielanie do iného inštancie NiFi. Prenos dát prebieha pomocou protokolu NiFi Site to Site.

NiFi Site to Site (S2S) — bezpečný, ľahko nastaviteľný spôsob prenosu dát medzi inštanciami NiFi. Ako S2S funguje, pozrite sa v dokumenácie a je dôležité nezabudnúť nastaviť inštanciu NiFi tak, aby povolila S2S, pozrite sa tu.

V prípadoch, keď ide o prenos dát pomocou S2S — jedna inštancia sa nazýva klientská, druhá serverová. Klientská odosiela dáta, serverová ich prijíma. Existujú dva spôsoby, ako nastaviť prenos dát medzi nimi:

  1. Push. Z klientskej inštancie sú dáta odosielané pomocou Remote Process Group (RPG). Na serverovej inštancii sú dáta prijímané pomocou Input Port
  2. Pull. Server prijíma dáta pomocou RPG, klient odosiela pomocou Output port.


Flow na nasadenie uchovávame v Apache Registry.

Apache NiFi Registry je podprojekt Apache NiFi, ktorý predstavuje nástroj na uchovávanie flow a správu verzií. Niečo ako GIT. Informácie o inštalácii, nastavení a práci s registry nájdete v oficiálnej dokumentácii. Flow na uchovávanie je spojené do procesnej skupiny a v takej podobe sa uchováva v registry. Následne sa k tomu v článku ešte vrátime.

Na začiatku, keď je N malé číslo, flow sa dodáva a aktualizuje ručne za prijateľný čas.

Ale so zvyšovaním N sa problémov objavuje viac:

  1. na aktualizáciu flow sa vynakladá viac času. Treba sa prihlásiť na všetky servery
  2. objavujú sa chyby pri aktualizácii šablón. Tu sme aktualizovali, ale tu sme zabudli
  3. chyby človeka pri vykonávaní veľkého počtu homogénnych operácií

To všetko nás vedie k tomu, že proces treba automatizovať. Skúšal som nasledujúce spôsoby riešenia tejto úlohy:

  1. Používať MiNiFi namiesto NiFi
  2. NiFi CLI
  3. NiPyAPI

Použitie MiNiFi

Apache MiNiFy — podprojekt Apache NiFi. MiNiFy — kompaktný agent, ktorý používa tie isté procesory ako NiFi, čo umožňuje vytvárať rovnaké flow ako v NiFi. Ľahkosť agenta je dosiahnutá aj tým, že MiNiFy nemá grafické rozhranie na konfiguráciu flow. Absencia grafického rozhrania v MiNiFy znamená, že je potrebné riešiť problém dodania flow do minifi. Keďže MiNiFy sa aktívne používa v IOT, je to komplexné a proces dodania flow k konečným inštanciám minifi treba automatizovať. Známa úloha, však?

Riešenie tejto úlohy môže poskytnúť ďalší podprojekt — MiNiFi C2 Server. Tento produkt je navrhnutý na to, aby bol centrálnym bodom v architektúre nasadzovania konfigurácií. Ako nakonfigurovať prostredie — je popísané v tomto článku na Habré a informácií je dosť na vyriešenie danej úlohy. MiNiFi v spojení s C2 serverom automaticky aktualizuje svoju konfiguráciu. Jedinou nevýhodou tohto prístupu je, že je potrebné vytvárať šablóny na C2 Server, 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 treba nezabudnúť 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 tejto 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údajúc podľa popisu nástroja na oficiálnej stránke, ide o nástroj na automatizáciu interakcie NiFi a NiFi Registry v oblasti dodania flow alebo správy procesov. Na začiatok práce je potrebné tento nástroj stiahnuť. odtiaľto.

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 mohli načítať potrebný flow z registry, musíme poznať identifikátory vedra (bucket identifier) a samotného flow (flow identifier). Tieto údaje môžeme získať buď prostredníctvom cli, alebo vo webovom rozhraní NiFi registry. Vo webovom rozhraní to vyzerá takto:

Automatizácia dodávky flow v Apache NiFi

Pomocou CLI to vyzerá 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 process group 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

Dôležitý bod — ako hostiteľ, na ktorý nasadzujeme process group, môže byť uvedený akýkoľvek inštancia nifi.

Process group bola pridaná s pozastavenými procesormi, musia sa spustiť.

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

Výborne, procesory sa spustili. Avšak, podľa podmienok úlohy musia inštancie NiFi posielať údaje na iné inštancie. Predpokladajme, že na prenos údajov na server bol zvolený spôsob Push. Pre organizovanie prenosu údajov musíme na pridanom Remote Process Group (RPG), ktorý už je zahrnutý v našom flow, povoliť prenos údajov (Enable transmitting).

Automatizácia dodávky flow v Apache NiFi

V dokumentácii v CLI a iných zdrojoch som nenašiel spôsob, ako povoliť prenos údajov. Ak viete, ako to urobiť — prosím, napíšte to do komentárov.

Keďže máme bash a sme pripravení ísť do konca — nájdeme východ! Môžeme využiť NiFi API na riešenie tohto problému. Použijeme nasledujúcu metódu, ID vezmeme z príkladov vyššie (v našom prípade je to 7f522a13-016e-1000-e504-d5b15587f2f3). Popis metód NiFi API. tu.

Automatizácia dodávky flow v Apache NiFi
Do body je potrebné odoslať JSON nasledujúceho typu:

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

Parametre, ktoré je potrebné vyplniť, aby „to fungovalo“:
state — stav prenosu údajov. K dispozícii je TRANSMITTING na povolenie prenosu údajov, STOPPED na jeho vypnutie.
version — verzia procesora.

verzia bude predvolene 0 pri vytvorení, ale tieto parametre je možné získať pomocou metódy.

Automatizácia dodávky flow v Apache NiFi

Pre milovníkov bash skriptov môže byť táto metóda zaujímavá, ale pre mňa je to ťažké — bash skripty nie sú mojím obľúbeným. Nasledujúci spôsob je podľa môjho názoru zaujímavejší a pohodlnejší.

NiPyAPI

NiPyAPI — knižnica pre jazyk Python na interakciu s inštanciami NiFi. Stránka s dokumentáciou obsahuje potrebné informácie na prácu s knižnicou. Rýchly štart je popísaný v projekte na github.

Náš skript na nasadenie konfigurácie — program v jazyku Python. Prejdeme k programovaniu.
Nastavujeme konfigurácie na ďalšiu prácu. Budeme potrebovať nasledujúce parametre:

nipyapi.config.nifi_config.host = 'http://nifi:8080/nifi-api' #cesta k nifi-api inštancii, na ktorej nasadzujeme process group
nipyapi.config.registry_config.host = 'http://nifi-registry:18080/nifi-registry-api' #cesta k nifi-registry-api registry
nipyapi.config.registry_name = 'MyBeautifulRegistry' #názov registry, ako sa bude nazývať v inštancii nifi
nipyapi.config.bucket_name = 'BucketName' #názov bucket, z ktorého ťaháme flow
nipyapi.config.flow_name = 'FlowName' #názov flow, ktorý ťaháme

Ďalej budem vkladať názvy metód tejto knižnice, ktoré sú popísané tu.

Pripájame registry k inštancii nifi pomocou

nipyapi.versioning.create_registry_client

Na tomto kroku môžeme ešte pridať kontrolu, či je registry už k inštancii pridaný, na to môžeme využiť metódu

nipyapi.versioning.list_registry_clients

Nájdeme bucket na ďalšie hľadanie flow v koši

nipyapi.versioning.get_registry_bucket

Podľa nájdeného bucketu hľadáme flow

nipyapi.versioning.get_flow_in_bucket

Ďalej je dôležité pochopiť, či už bol tento process group pridaný. Process group sa umiestňuje podľa súradníc a môže nastat situácia, keď sa na jeden komponent nanesie druhý. Overil som, môže to byť 🙂. Aby sme získali všetky pridané process groups, použijeme metódu

nipyapi.canvas.list_all_process_groups

a potom môžeme hľadať, napríklad, podľa mena.

Nebudem opisovať proces aktualizácie šablóny, poviem iba to, že ak sa v novej verzii šablóny pridávajú procesory, tak nie sú problémy s prítomnosťou správ v frontách. Ale ak sa procesory odstraňujú, môžu vzniknúť problémy (nifi nedovolí odstrániť procesor, ak sa pred ním nahromadila fronta správ). Ak vás zaujíma, ako som tento problém vyriešil — napíšte mi prosím, preberieme tento moment. Kontakty na konci článku. Prejdime k kroku pridania process group.

Pri ladení skriptu som sa stretol s zvláštnosťou, že nie vždy sa stiahne najnovšia verzia flow, preto odporúčam najprv túto verziu presne určiť:

nipyapi.versioning.get_latest_flow_ver

Nasadzujeme process group:

nipyapi.versioning.deploy_flow_version

Spúšťame procesory:

nipyapi.canvas.schedule_process_group

V bloku o CLI bolo napísané, že v remote process group sa automaticky nezapína prenos dát? Pri realizácii skriptu som sa s touto problémom tiež stretol. V tom čase sa mi nepodarilo spustiť prenos dát pomocou API a rozhodol som sa napísať developerovi knižnice NiPyAPI a požiadať ho o radu/pomoc. Developer mi odpovedal, prebrali sme problém a povedal, že potrebuje čas „skontrolovať niečo“. A tak, po pár dňoch prišiel e-mail, v ktorom bola funkcia na Pythone, ktorá riešila môj problém spustenia!!! V tom čase bola verzia NiPyAPI 0.13.3 a v nej, samozrejme, nič také nebolo. Ale vo verzii 0.14.0, ktorá bola vydaná prednedávnom, táto funkcia už bola súčasťou knižnice. Vitajte,

nipyapi.canvas.set_remote_process_group_transmission

Takže, s pomocou knižnice NiPyAPI sme pripojili registry, nasadili flow a dokonca spustili procesory a prenos dát. Ďalej môžeme doladiť kód, pridať rôzne kontroly, logovanie a to všetko. Ale to je už úplne iný príbeh.

Z preverených možností automatizácie sa mi posledná zdala najviac funkčná. Po prvé, je to predsa len kód v Pythone, do ktorého môžeme integrovať pomocný programový kód a využiť všetky výhody programovacieho jazyka. Po druhé, projekt NiPyAPI sa aktívne vyvíja a v prípade problémov môžeme napísať vývojárovi. Po tretie, NiPyAPI je predsa len flexibilnejší nástroj na interakciu s NiFi pri riešení zložitých úloh. Napríklad pri zisťovaní, či sú aktuálne fronty správ prázdne a či je možné aktualizovať process group.

Na tomto konci. Opísal som 3 prístupy k automatizácii dodávky flow do NiFi, úskalia, s ktorými sa môže vývojár stretnúť, a poskytol pracovný kód na automatizáciu dodávky. Ak vás táto téma tiež zaujíma, rovnako ako mňa — napíšte!

Zdroj: habr.com

Kúpte si spoľahlivý hosting pre webové stránky s ochranou proti DDoS, VPS VDS servery 🔥 Kúpte si spoľahlivý hosting pre webové stránky s ochranou proti DDoS, VPS VDS servery | ProHoster