Ahoj všichni!

Úkolem je následující — existuje flow, který je představen na obrázku výše, který je potřeba nasadit na N serverech s Apache NiFi. Flow je testovací — probíhá generování souboru a jeho odesílání do jiného instance NiFi. Předávání dat probíhá pomocí protokolu NiFi Site to Site.

NiFi Site to Site (S2S) — bezpečný, snadno konfigurovatelný způsob přenosu dat mezi instancemi NiFi. Jak S2S funguje, můžete vidět v dokumentaci a je důležité nezapomenout nastavit instanci NiFi, aby povolila S2S, jak vidíte tady.

V případech, kdy jde o přenos dat pomocí S2S — jedna instance se nazývá klientská, druhá serverová. Klient odesílá data, server přijímá. Existují dva způsoby, jak nastavit přenos dat mezi nimi:

1. Push. Z klientské instance jsou data odesílána pomocí Remote Process Group (RPG). Na serverové instanci jsou data přijímána pomocí Input Port
2. Pull. Server přijímá data přes RPG, klient odesílá pomocí Output port.

Flow pro nasazení uchováváme v Apache Registry.

Apache NiFi Registry — podprojekt Apache NiFi, který představuje nástroj pro uchovávání flow a správu verzí. Něco jako GIT. Informace o instalaci, konfiguraci a práci s registry naleznete v oficiální dokumentaci. Flow pro uchovávání je sdruženo do process group a v této podobě je uchováváno v registry. K tomuto tématu se v článku ještě vrátíme.

Na začátku, když je N malé číslo, je flow dodáváno a aktualizováno ručně v přijatelném čase.

Ale s rostoucím N se problémů objevuje více:

1. aktualizace flow trvá více času. Je třeba se přihlásit na všechny servery
2. objevují se chyby v aktualizaci šablon. Tady bylo aktualizováno, ale zde jsme zapomněli
3. chybami člověka při provádění velkého množství podobných operací

To nás přivádí k závěru, že je potřeba proces automatizovat. Zkoušel jsem následující způsoby, jak tento úkol vyřešit:

1. Použít MiNiFi místo NiFi
2. NiFi CLI
3. NiPyAPI

Použití MiNiFi

Apache MiNiFy — subprojekt Apache NiFi. MiNiFy — kompaktní agent, který využívá stejné procesory jako NiFi a umožňuje vytvářet stejné flow jako v NiFi. Lehkost agenta se dosahuje také tím, že MiNiFy nemá grafické rozhraní pro konfiguraci flow. Absence grafického rozhraní u MiNiFy znamená, že je třeba řešit problém doručování flow do minifi. Vzhledem k tomu, že MiNiFy se aktivně používá v IOT, je komponentů mnoho a proces doručování flow k cílovým instancím minifi je třeba automatizovat. Známa úloha, že?

Vyřešit takový úkol pomůže další subprojekt — MiNiFi C2 Server. Tento produkt je určen k tomu, aby byl centrálním bodem v architektuře nasazení konfigurací. Jak konfigurovat prostředí — je popsáno v tomto článku na Habr a informací je dostatek pro řešení zadaného úkolu. MiNiFi ve spojení s C2 serverem automaticky aktualizuje konfiguraci u sebe. Jedinou nevýhodou tohoto přístupu je potřeba vytvářet šablony na C2 Server, jednoduché komitování do registry nestačí.

Možnost popsaná v článku výše je funkční a není obtížná na realizaci, ale je třeba nezapomínat na následující:

1. V minifi nejsou všechny procesory z nifi.
2. Verze procesorů v Minifi zaostávají za verzemi procesorů v NiFi.

V době psaní publikace byla poslední verze NiFi — 1.9.2. Verze procesorů poslední verze MiNiFi — 1.7.0. Procesory lze přidávat do MiNiFi, ale kvůli rozdílům verzí mezi procesory NiFi a MiNiFi to nemusí fungovat.

NiFi CLI

S ohledem na popisu nástroje na oficiálních stránkách, je to nástroj pro automatizaci interakce NiFi a NiFi Registry v oblasti doručování flow nebo řízení procesů. Pro zahájení práce je třeba si tento nástroj stáhnout. odsud.

Spustíme utilitu

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

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

Abychom mohli načíst potřebné flow z registry, musíme znát identifikátory koše (bucket identifier) a samotného flow (flow identifier). Tyto údaje lze získat buď přes cli, nebo v webovém rozhraní NiFi registry. Ve webovém rozhraní to vypadá takto:

Pomocí CLI se to dělá 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

Spustíme 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 — jako hostitel, na který nasazujeme process group, může být uveden jakýkoli instance nifi.

Skupina procesů byla přidána se zastavenými procesory, je třeba je spustit

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

Skvěle, procesory se spustily. Nicméně, podle podmínek úkolu musíme zajistit, aby instance NiFi posílaly data na jiné instance. Předpokládejme, že jsme pro přenos dat na server vybrali způsob Push. Abychom zorganizovali přenos dat, musíme na přidané skupině vzdálených procesů (RPG), která je již zahrnuta v našem toku, aktivovat přenos dat (Enable transmitting).

V dokumentaci v CLI a dalších zdrojích jsem nenašel způsob, jak povolit přenos dat. Pokud víte, jak to udělat – prosím, napište to do komentářů.

Když máme bash a jsme připraveni to dotáhnout do konce – najdeme řešení! Můžeme využít NiFi API k vyřešení tohoto problému. Využijeme následující metodu, ID vezmeme z výše uvedených příkladů (v našem případě je to 7f522a13-016e-1000-e504-d5b15587f2f3). Popis metod NiFi API tady.

V těle je třeba poslat JSON následujícího typu:

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

Parametry, které je třeba vyplnit, aby to "fungovalo":
stav — stav přenosu dat. K dispozici TRANSMITTING pro povolení přenosu dat, STOPPED pro vypnutí
version — verze procesoru

verze nastavení bude 0 při vytvoření, ale tyto parametry lze získat pomocí metody

Pro milovníky bash skriptů se může tato metoda zdát použitelná, ale mě to moc nevyhovuje – bash skripty nejsou moje nejoblíbenější. Další způsob je podle mě zajímavější a pohodlnější.

NiPyAPI

NiPyAPI – knihovna pro jazyk Python pro interakci s instancemi NiFi. Stránka s dokumentací obsahuje potřebné informace pro práci s knihovnou. Rychlý start je popsán v projektu na github.

Náš skript pro nasazení konfigurace – program na jazyku Python. Přejděme k programování.
Nastavujeme konfigurace pro další práci. Budeme potřebovat následující parametry:

nipyapi.config.nifi_config.host = 'http://nifi:8080/nifi-api' # cesta k instanci nifi-api, na které nasazujeme skupinu procesů
nipyapi.config.registry_config.host = 'http://nifi-registry:18080/nifi-registry-api' # cesta k nifi-registry-api registry
nipyapi.config.registry_name = 'MyBeutifulRegistry' # název registry, jak bude pojmenována v instanci nifi
nipyapi.config.bucket_name = 'BucketName' # název bucketu, ze kterého stahujeme flow
nipyapi.config.flow_name = 'FlowName' # název flow, které stahujeme

Dále budu uvádět názvy metod této knihovny, které jsou popsány tady.

Připojujeme registry k instanci nifi pomocí

nipyapi.versioning.create_registry_client

V tomto kroku můžete také přidat kontrolu, zda je registr již přidán k instanci. K tomu můžete použít metodu

nipyapi.versioning.list_registry_clients

Najdeme bucket pro další vyhledávání flow v koši

nipyapi.versioning.get_registry_bucket

Podle nalezeného bucketu hledáme flow

nipyapi.versioning.get_flow_in_bucket

Dále je důležité pochopit, zda již není tento process group přidán. Process group je umístěn podle souřadnic a může se stát, že se na jeden komponent nanese druhý. Ověřoval jsem to, taková situace může nastat 🙂. Abychom získali všechny přidané process group, použijeme metodu

nipyapi.canvas.list_all_process_groups

a poté můžeme hledat například podle názvu.

Nebudu popisovat proces aktualizace šablony, řeknu jen, že pokud jsou ve nové verzi šablony přidány procesory, s přítomností zpráv v frontách problém není. Pokud jsou procesory odstraněny, mohou se vyskytnout problémy (nifi nedovoluje odstranit procesor, pokud se před ním nahromadila fronta zpráv). Pokud vás zajímá, jak jsem tento problém vyřešil, napište mi, prosím, probereme tento bod. Kontakty na konci článku. Přejdeme k kroku přidání process group.

Při ladění skriptu jsem se setkal s tím, že se ne vždy stahuje poslední verze flow, proto doporučuji nejprve tuto verzi upřesnit:

nipyapi.versioning.get_latest_flow_ver

Deployujeme process group:

nipyapi.versioning.deploy_flow_version

Spouštíme procesory:

nipyapi.canvas.schedule_process_group

V bloku o CLI bylo napsáno, že v remote process group se automaticky nezahrnuje přenos dat? Při implementaci skriptu jsem se s tímto problémem také setkal. V tu chvíli se mi nedařilo spustit přenos dat pomocí API, a tak jsem se rozhodl napsat vývojáři knihovny NiPyAPI a požádat o radu/pomoc. Vývojář mi odpověděl, my jsme probrali problém a on napsal, že potřebuje čas „ověřit něco“. A tak, po několika dnech, přišel e-mail, ve kterém byla funkce v Pythonu, řešící můj problém spuštění!!! Tehdy byla verze NiPyAPI 0.13.3 a samozřejmě v ní nic takového nebylo. Ale ve verzi 0.14.0, která byla vydána nedávno, již tato funkce byla součástí knihovny. Vítejte,

nipyapi.canvas.set_remote_process_group_transmission

Takže pomocí knihovny NiPyAPI jsme se připojili k registru, zavřeli flow a dokonce spustili procesory a přenos dat. Teď můžeme upravit kód, přidat různé kontroly, logování a podobně. Ale to už je úplně jiný příběh.

Z možností automatizace, které jsem zvažoval, se mi poslední zdál nejpraktičtější. Za prvé, je to přece jen kód v Pythonu, do kterého lze začlenit pomocný programový kód a využívat všechny výhody programovacího jazyka. Za druhé, projekt NiPyAPI se aktivně vyvíjí a v případě problémů lze napsat vývojáři. Za třetí, NiPyAPI je přece jen flexibilnější nástroj pro interakci s NiFi při řešení složitých úloh. Například při určování, zda jsou fronty zpráv nyní v flow prázdné a zda lze aktualizovat process group.

To je vše. Popsal jsem 3 přístupy k automatizaci dodávky flow do NiFi, nástrahy, na které může vývojář narazit, a uvedl funkční kód pro automatizaci dodávky. Pokud vás toto téma zajímá stejně jako mě — napište!

Zdroj: habr.com