Përshëndetje të gjithëve!

Detyra është si vijon – ka një fluks, i paraqitur në figurën e mësipërme, që duhet të implementohet në N serverë me Apache NiFi. Fluksi është testues – duke krijuar një skedë dhe duke e dërguar në një instancë tjetër të NiFi. Shkëmbimi i të dhënave bëhet nëpërmjet protokollit NiFi Site to Site.

NiFi Site to Site (S2S) – një mënyrë e sigurtë dhe e lehtë për t'u konfiguruar për transfertën e të dhënave midis instancave NiFi. Si funksionon S2S shihni në dokumentacionin dhe është e rëndësishme të mos harrosh të konfigurash instancën NiFi që të lejojë S2S, shihni këtu.

Në rastet kur bëhet fjalë për transfertën e të dhënave me S2S – një instancë quhet klient, tjetra server. Klienti dërgon të dhënat, serveri i merr ato. Janë dy mënyra për të konfiguruar transfertën e të dhënave midis tyre:

1. Shtytje. Nga instanca klient, të dhënat dërgohen përmes Remote Process Group (RPG). Në instancën server të dhënat priten përmes Input Port.
2. Tërheqje. Serveri pranën të dhënat përmes RPG, klienti dërgon përmes Output port.

Fluksi për implementimin e mban në Apache Registry.

Apache NiFi Registry – një nënprojekt i Apache NiFi, që paraqet një mjet për ruajtjen e flukseve dhe menaxhimin e versioneve. Një lloj GIT-i. Informacionin mbi instalimin, konfigurimin dhe punën me registry mund ta gjeni në dokumentacionin zyrtar. Flukset për ruajtje bashkohen në një grup procesesh dhe kështu ruhet në registry. Më vonë në artikull do të kthehemi përsëri në këtë temë.

Në fillim, kur N është një numër i vogël, fluksi dërgohet dhe përditësohet manualisht brenda një kohe të pranueshme.

Por me rritjen e N, problemet bëhen më të numra:

1. në përditësimin e fluksit shpenzohet më shumë kohë. Duhet të hysh në të gjitha serverët
2. kanë ndodhur gabime në përditësimin e shablloneve. Këtu është përditësuar, por këtu është harruar
3. gabime njerëzore kur kryhen shumë operacione të ngjashme

Të gjitha këto na çojnë në konstatimin se është e nevojshme automatizimi i procesit. Kam provuar mënyra të tjera për të zgjidhur këtë detyrë:

1. Të përdorim MiNiFi në vend të NiFi
2. NiFi CLI
3. NiPyAPI

Përdorimi i MiNiFi

Apache MiNiFy — projekt anësor Apache NiFi. MiNiFy — një agent kompakt që përdor të njëjtat processorë si NiFi, duke mundësuar krijimin e të njëjtave flow si në NiFi. Lehtësia e agentit arrihet gjithashtu për shkak se MiNiFy nuk ka një ndërfaqe grafike për konfigurimin e flow. Mungesa e një ndërfaqe grafike në MiNiFy do të thotë se duhet të zgjidhni problemin e dërgimit të flow në minifi. Passi, MiNiFy përdoret aktivisht në IOT, komponentët janë shumë dhe procesi i dërgimit të flow deri te instancat e fundit minifi duhet të automatizohet. Një detyrë e njohur, apo jo?

Zgjidhja e një detyre të tillë do të ndihmojë një projekt anësor tjetër — Serveri MiNiFi C2. Ky produkt është i destinuar për të qenë pika qendrore në arkitekturën e shpërndarjes së konfigurimeve. Si të konfigurohet mjedisi — është përshkruar në këto artikuj në Habr dhe informacioni është mjaftueshëm për të zgjidhur detyrën e vendosur. MiNiFi në bashkëpunim me serverin C2 përditëson automatikisht konfigurimin e tij. E vetmja mangësi e këtij qasjeje është se duhet të krijoni shabllone në Serverin C2, një komit i thjeshtë në registry nuk është i mjaftueshëm.

Opcioni i përshkruar në artikullin e mësipërm është funksional dhe i thjeshtë për t'u realizuar, por nuk duhet të harrojmë sa vijon:

1. Në minifi nuk janë të gjitha processorët e nifi
2. Versionet e processorëve në Minifi po vonojnë nga versionet e processorëve në NiFi.

Në momentin e shkruarjes së publikimit, versioni më i fundit i NiFi është 1.9.2. Versioni i processorëve të versionit më të fundit të MiNiFi është 1.7.0. Processorët mund të shtohen në MiNiFi, por për shkak të dallimeve në versionet midis processorëve NiFi dhe MiNiFi mund të mos funksionojë.

NiFi CLI

Sipas përshkrimit të mjetit në faqen zyrtare, ky është një mjet për automatizimin e ndërveprimit të NiFI dhe NiFi Registry në fushën e dërgimit të flow ose menaxhimit të proceseve. Për të filluar punën, ky mjet duhet të shkarkohet këtu.

E aktivizojmë utilitarin

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

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

Për të ngarkuar flow të nevojshëm nga registry, na nevojitet të dimë identifikuesit e koshit (bucket identifier) dhe të vetë flow (flow identifier). Këto të dhëna mund të merren përmes cli ose në ndërfaqen e uebit të NiFi registry. Në ndërfaqen e uebit duket kështu:

Me CLI bëhet kështu:

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

E aktivizojmë importimin e grupit të proceseve nga 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

Një pikë e rëndësishme është se si host, mbi të cilin ne vendosim grupin e proceseve, mund të zbulohet çdo instancë nifi.

Grupi i procesit është shtuar me procesorë të ndalur, ata duhet të nisën.

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

Shumë mirë, procesorët filluan. Megjithatë, sipas kushteve të detyrës, ne duhet që instancat NiFi të dërgojnë të dhëna në instanca të tjera. Le të supozojmë se për transmetimin e të dhënave në server është zgjedhur metoda Push. Për të organizuar transmetimin e të dhënave, është e nevojshme që në Grupin e Procesit të Largët (RPG), i cili tashmë është i aktivizuar në rrjedhën tonë, të aktivizoni transmetimin e të dhënave (Enable transmitting).

Në dokumentacionin e CLI dhe burime të tjera nuk kam gjetur një mënyrë për të aktivizuar transmetimin e të dhënave. Nëse e dini se si ta bëni këtë, ju lutem shkruani në komentet.

Duke pasur parasysh se kemi bash dhe jemi të gatshëm të shkojmë deri në fund - le të gjejmë një zgjidhje! Mund të përdorim API-në e NiFi për të zgjidhur këtë problem. Do ta përdorim metodën e mëposhtme, ID-në e marrim nga shembujt më sipër (në rastin tonë ajo është 7f522a13-016e-1000-e504-d5b15587f2f3). Përshkrimi i metodave të API-së së NiFi. këtu.

Në body duhet të dërgohet JSON, i tillë si në vijim:

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

Parametrat që duhet të plotësohen për ta bërë "të funksionojë":
state — statusi i transmetimit të të dhënave. Disponohet TRANSMITTING për aktivizimin e transmetimit të të dhënave, STOPPED për ndërprerjen.
version — versioni i procesorit.

Versioni në mënyrë standarde do të jetë 0 në krijim, por këta parametra mund të merren duke përdorur metodën.

Për ata që pëlqejnë skriptet bash, kjo metodë mund të duket e përshtatshme, por mua më duket e vështirë - skriptet bash nuk janë gjëja ime më e preferuar. Mënyra tjetër është më interesante dhe më e përshtatshme sipas mendimit tim.

NiPyAPI

NiPyAPI - një bibliotekë për gjuhën Python për të ndërvepruar me instancat NiFi. Faqja me dokumentacionin përmban informacionin e nevojshëm për punën me bibliotekën. Shkurtimi fillestar është përshkruar në projektin në github.

Skripti ynë për shpërndarjen e konfiguracionit është një program i shkruar në Python. Të kalojmë në kodimin.
Përgatitja e konfigurimeve për punë më të vonshme. Na nevojiten parametrat e mëposhtëm:

nipyapi.config.nifi_config.host = 'http://nifi:8080/nifi-api' #rruga deri te instanca nifi-api, mbi të cilën po zhvillojmë grupin e proceseve
nipyapi.config.registry_config.host = 'http://nifi-registry:18080/nifi-registry-api' #rruga deri te nifi-registry-api registry
nipyapi.config.registry_name = 'MyBeutifulRegistry' #emri i registry-t, si do të quhet në instancën nifi
nipyapi.config.bucket_name = 'BucketName' #emri i bucket-it, nga i cili tërheqim rrjedhën
nipyapi.config.flow_name = 'FlowName' #emri i rrjedhës, që po tërheqim

Më tej do të përfshij emrat e metodave të kësaj biblioteke, që janë të përshkruara këtu.

Lidhim regjistrin me instancën nifi duke përdorur

nipyapi.versioning.create_registry_client

Në këtë hap mund të shtojmë gjithashtu një kontroll nëse regjistri është tashmë i shtuar në instancë; për këtë mund të përdorim metodën

nipyapi.versioning.list_registry_clients

Gjejmë bucket për kërkimin e mëtejshëm flow në kosh

nipyapi.versioning.get_registry_bucket

Nga bucket-i i gjetur kërkojmë flow

nipyapi.versioning.get_flow_in_bucket

Më pas është e rëndësishme të kuptojmë nëse ky process group është shtuar tashmë. Process group pozicionohet sipas koordinatave dhe mund të ndodhë që një komponent të mbivendoset mbi një tjetër. E kam verifikuar, ndodh kështu 🙂 Për të marrë të gjitha process group që janë të shtuar, përdorim metodën

nipyapi.canvas.list_all_process_groups

dhe më pas mund të kërkojmë, për shembull, sipas emrit.

Nuk do të përshkruaj procesin e përditësimit të templatit, vetëm do të them se nëse në versionin e ri të templatit shtohen procesorë, nuk ka probleme me praninë e mesazheve në radhë. Por nëse procesorë hiqen, mund të lindin probleme (nifi nuk lejon heqjen e një procesori nëse përpara tij ka rënë një radhë mesazhesh). Nëse jeni të interesuar si e zgjidh këtë problem — ju lutem, më shkruani, do të diskutojmë këtë çështje. Kontaktet janë në fund të artikullit. Të kalojmë në hapin e shtimit të process group.

Gjatë debug-ut të skriptit u ndesha me një veçori që nuk gjithmonë arrin të tërheqë versionin më të fundit të flow, prandaj rekomandoj së pari ta sqaroni këtë version:

nipyapi.versioning.get_latest_flow_ver

Deploy-ojmë process group:

nipyapi.versioning.deploy_flow_version

Kemi filluar procesorët:

nipyapi.canvas.schedule_process_group

Në bllokun për CLI u shkrua se dërgimi i të dhënave në remote process group nuk aktivizohet automatikisht? Gjatë realizimit të skriptit, ndesha këtë problem gjithashtu. Në atë kohë, nuk arrita të aktivizoj dërgimin e të dhënave përmes API-së dhe vendosa të shkruaj zhvilluesit të bibliotekës NiPyAPI dhe të kërkoj ndihmë/këshillë. Zhvilluesi më përgjigji, ne diskutuan problemin dhe ai shkroi se i nevojitej një kohë “për të kontrolluar diçka”. Dhe ja, pas disa ditësh, vjen një letër ku është e shkruar një funksion në Python që zgjidh problemin tim të aktivizimit!!! Në atë kohë, versioni i NiPyAPI ishte 0.13.3 dhe në të, natyrisht, nuk kishte asgjë të tillë. Por në versionin 0.14.0, që doli së fundmi, ky funksion është përfshirë në bibliotekë. Mirë se erdhët,

nipyapi.canvas.set_remote_process_group_transmission

Pra, duke përdorur bibliotekën NiPyAPI, lidhëm regjistrin, vendosëm flow dhe madje aktivizuam procesorët dhe transferimin e të dhënave. Më pas, mund të rregullojmë kodin, të shtojmë kontrollime të ndryshme, regjistrimin dhe gjithë këtë. Por kjo është një histori krejt tjetër.

Nga opsionet e automatizimit që shqyrtova, më fundi më duket se është më funksionale. Së pari, prapë kemi kod në python, në të cilin mund të integrojmë kod ndihmës dhe të shfrytëzojmë të gjitha avantazhet e gjuhës së programimit. Së dyti, projekti NiPyAPI po zhvillohet aktivisht dhe në rast problemesh mund të shkruajmë zhvilluesit. Së treti, NiPyAPI është një mjet më fleksibël për ndërveprimin me NiFi në zgjidhjen e detyrave komplekse. Për shembull, në përcaktimin nëse radhët e mesazheve janë bosh tani në flow dhe nëse mund të azhurnojmë grupin e proceseve.

Këtu përfundon. Kam përshkruar 3 qasje për automatizimin e dorëzimit të flow në NiFi, pengesat me të cilat mund të përballet zhvilluesi dhe kam dhënë kod funksional për automatizimin e dorëzimit. Nëse ju intereson kjo temë ashtu si edhe mua — shkruani!

Burimi: habr.com