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 . 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Ă« dhe Ă«shtĂ« e rĂ«ndĂ«sishme tĂ« mos harrosh tĂ« konfigurash instancĂ«n NiFi qĂ« tĂ« lejojĂ« S2S, shihni .
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:
- 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.
- 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Ă« . 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:
- në përditësimin e fluksit shpenzohet më shumë kohë. Duhet të hysh në të gjitha serverët
- kanë ndodhur gabime në përditësimin e shablloneve. Këtu është përditësuar, por këtu është harruar
- 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ë:
- Të përdorim MiNiFi në vend të NiFi
- NiFi CLI
- NiPyAPI
PĂ«rdorimi i MiNiFi
â 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Ă« 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:
- Në minifi nuk janë të gjitha processorët e nifi
- 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 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 .
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. .
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. përmban informacionin e nevojshëm për punën me bibliotekën. Shkurtimi fillestar është përshkruar në 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 .
Lidhim regjistrin me instancën nifi duke përdorur
nipyapi.versioning.create_registry_clientNë 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_clientsGjejmë bucket për kërkimin e mëtejshëm flow në kosh
nipyapi.versioning.get_registry_bucketNga bucket-i i gjetur kërkojmë flow
nipyapi.versioning.get_flow_in_bucketMĂ« 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_groupsdhe 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_verDeploy-ojmë process group:
nipyapi.versioning.deploy_flow_versionKemi filluar procesorët:
nipyapi.canvas.schedule_process_groupNĂ« 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_transmissionPra, 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 â
Burimi: habr.com
