Përshëndetje të gjithëve!
Detyra Ă«shtĂ« si vijon â ka njĂ« flow, i paraqitur nĂ« imazhin pĂ«rmĂ«nshĂ«m, qĂ« duhet tĂ« shpĂ«rndahet nĂ« N servera me . Flow Ă«shtĂ« testues â ndodhet gjenerimi i njĂ« skedari dhe dĂ«rgimi nĂ« njĂ« instancĂ« tjetĂ«r NiFi. Transferimi i tĂ« dhĂ«nave ndodh pĂ«rmes protokollit NiFi Site to Site.
NiFi Site to Site (S2S) â njĂ« mĂ«nyrĂ« e sigurt, e lehtĂ« pĂ«r t'u konfiguruar pĂ«r transferimin e tĂ« dhĂ«nave mes instancave NiFi. Si funksionon S2S mund tĂ« shikoni nĂ« dhe Ă«shtĂ« e rĂ«ndĂ«sishme tĂ« mos harrohet tĂ« konfigurohet instanca NiFi pĂ«r tĂ« lejuar S2S, shihni .
NĂ« rastet kur flitet pĂ«r transferimin e tĂ« dhĂ«nave pĂ«rmes S2S â njĂ« instancĂ« quhet klient, ndĂ«rsa tjetra shĂ«rbim. Klienti dĂ«rgon tĂ« dhĂ«na, serveri i pranon. JanĂ« dy mĂ«nyra pĂ«r tĂ« konfiguruar transferimin e tĂ« dhĂ«nave mes tyre:
- Push. Nga instanca klient të dhënat dërgohen përmes Remote Process Group (RPG). Në instancën server të dhënat pranohen përmes Input Port
- Tërheqje. Serveri pranon të dhëna përmes RPG, klienti dërgon përmes Output port.
Flow për shpërndarje e ruajmë në Apache Registry.
Apache NiFi Registry â nĂ«nprojekte Apache NiFi, qĂ« ofron njĂ« mjet pĂ«r ruajtjen e flukseve dhe menaxhimin e versioneve. NjĂ« lloj GIT-i. Informacione rreth instalimit, konfigurimit dhe pĂ«rdorimit tĂ« registry mund tĂ« gjenden nĂ« . Flukset pĂ«r ruajtje bashkohen nĂ« njĂ« grup procesesh dhe ruhen nĂ« atĂ« formĂ« nĂ« registry. MĂ« vonĂ« nĂ« kĂ«tĂ« artikull do tĂ« kthehemi sĂ«rish te kjo temĂ«.
Në fillim, kur N është një numër i vogël, fluksi dërgohet dhe përditësohet manualisht në një kohë të pranueshme.
Por me rritjen e N, problemet bëhen më shumë:
- përditësimi i fluksit kërkon më shumë kohë. Duhet të hysh në të gjitha serverat
- ndodhin gabime në përditësimin e shabloneve. Këtu e kemi përditësuar, por atje e kemi harruar
- gabimet e njeriut gjatë kryerjes së një numri të madh operacionesh të njëjta
Të gjitha këto na çojnë në përfundimin se duhet të automatizojmë procesin. Unë kam provuar këto mënyra për të zgjidhur këtë problem:
- Përdorimi i MiNiFi në vend të NiFi
- NiFi CLI
- NiPyAPI
PĂ«rdorimi i MiNiFi
â nĂ«nprojekt i Apache NiFi. MiNiFy â njĂ« agent kompakt qĂ« pĂ«rdor tĂ« njĂ«jtat procesorĂ« si NiFi, duke lejuar krijimin e tĂ« njĂ«jtave rrjedha si nĂ« NiFi. LehtĂ«sia e agentit arrihet gjithashtu pĂ«r shkak se MiNiFy nuk ka njĂ« ndĂ«rfaqe grafike pĂ«r konfigurimin e rrjedhave. Mungesa e ndĂ«rfaqes grafike nĂ« MiNiFy do tĂ« thotĂ« se duhet tĂ« zgjidhet problemi i dorĂ«zimit tĂ« rrjedhave nĂ« minifi. Duke qenĂ« se MiNiFy Ă«shtĂ« pĂ«rdorur gjerĂ«sisht nĂ« IOT, komponentĂ«t janĂ« tĂ« shumtĂ« dhe procesi i dorĂ«zimit tĂ« rrjedhave tek instancat pĂ«rfundimtare tĂ« minifi duhet tĂ« automatizohet. NjĂ« detyrĂ« e njohur, apo jo?
PĂ«r tĂ« zgjidhur njĂ« detyrĂ« tĂ« tillĂ« do tĂ« ndihmojĂ« njĂ« tjetĂ«r nĂ«nprojekt â MiNiFi C2 Server. Ky produkt Ă«shtĂ« menduar tĂ« jetĂ« pika qendrore nĂ« arkitekturĂ«n e shpĂ«rndarjes sĂ« konfigurimeve. Si tĂ« konfiguroni ambientin â Ă«shtĂ« pĂ«rshkruar nĂ« nĂ« Habr dhe informacioni Ă«shtĂ« i mjaftueshĂ«m pĂ«r tĂ« zgjidhur problemin e caktuar. MiNiFi nĂ« lidhje me serverin C2 pĂ«rditĂ«son automatikisht konfigurimin e tij. Disavantazhi i vetĂ«m i kĂ«tij qasjeje Ă«shtĂ« se duhet tĂ« krijoni shabllone nĂ« C2 Server, njĂ« angazhim i thjeshtĂ« nĂ« regjistrin nuk Ă«shtĂ« i mjaftueshĂ«m.
Në variantin e përshkruar në artikullin më sipër është funksional dhe jo i komplikuar për t'u zbatuar, por nuk duhet të harrojmë këtë:
- Në minifi nuk ka të gjithë procesorët nga nifi
- Versionet e procesorëve në Minifi mbeten pas atyre në NiFi.
Në momentin e publikimit, versioni më i fundit i NiFi është 1.9.2. Versioni i procesorëve të fundit të MiNiFi është 1.7.0. Procesorët mund të shtohen në MiNiFi, por për shkak të diferencave të versioneve mes procesorëve të NiFi dhe MiNiFi, kjo mund të mos funksionojë.
NiFi CLI
Duke u bazuar në të mjetit në faqen zyrtare, ky është një mjet për automatizimin e ndërveprimit mes NiFi dhe NiFi Registry në fushën e dorëzimit të flow ose menaxhimit të proceseve. Për të filluar punën, duhen shkarkuar këtë mjet .
SĂ« pari nisemi me utilitarin
./bin/cli.sh
_ ___ _
Apache (_) .' ..](_) ,
_ .--. __ _| |_ __ )
[ `.-. | [ |'-| |-'[ | /
| | | | | | | | | | ' '
[___||__][___][___] [___]', ,'
`'
CLI v1.9.2
Shkruani 'help' për të parë një listë të komandave të disponueshme, përdorni tab për të përfunduar automatikisht.
Për të importuar flow-in e nevojshëm nga registry, na nevojiten identifikuesit e koshes (bucket identifier) dhe të vetë flow-it (flow identifier). Këto të dhëna mund të merren ose përmes cli, ose në ndërfaqen web të NiFi registry. Në ndërfaqen web, duket kështu:
Me anë të 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
Nisemi me 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, ne mund të specifikojmë çdo instancë nifi nga procesi grup.
Grupi i proceseve është shtuar me procesorët e ndaluar, ata duhet të nisin.
#> nifi pg-start -pgid 7f522a13-016e-1000-e504-d5b15587f2f3 -u http://nifi:8080
Shumë mirë, procesorët filluan. Sidoqoftë, sipas kushteve të detyrës, 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, metoda e zgjedhur është Push. Për të organizuar transmetimin e të dhënave, duhet të aktivizojmë transmetimin e të dhënave në Remote Process Group (RPG) të shtuar, që tashmë është përfshirë në flow tonë.
Në dokumentacion, në 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.
Tani që kemi bash dhe jemi gati të shkojmë deri në fund, le të gjejmë një zgjidhje! Mund të përdorim NiFi API për të zgjidhur këtë problem. Do të përdorim metodën e mëposhtme, ID e marrim nga shembujt më sipër (në rastin tonë, kjo është 7f522a13-016e-1000-e504-d5b15587f2f3). Përshkrimi i metodave NiFi API. .
Në body duhet të dërgojmë JSON, në formën e mëposhtme:
{
"revision": {
"clientId": "value",
"version": 0,
"lastModifier": "value"
},
"state": "value",
"disconnectedNodeAcknowledged": true
}
Parametrat që duhet të plotësohen në mënyrë që "të funksionojë":
gjendja â statusi i transmetimit tĂ« tĂ« dhĂ«nave. E disponueshme TRANSMITTING pĂ«r aktivizimin e transmetimit tĂ« tĂ« dhĂ«nave, STOPPED pĂ«r ndalimin
version â versioni i procesorit
versioni në tërësi do të jetë 0 në krijim, por këto parametra mund të merren duke përdorur metodën
PĂ«r adhuruesit e skenareve bash, kjo metodĂ« mund tĂ« duket e dobishme, por Ă«shtĂ« pak e vĂ«shtirĂ« pĂ«r mua â skenaret bash nuk janĂ« gjĂ«ja ime mĂ« e preferuar. Metoda e ardhshme Ă«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Ă«rdor informacionin e nevojshĂ«m pĂ«r tĂ« punuar me bibliotekĂ«n. Shkurtimisht depozita Ă«shtĂ« e pĂ«rshkruar nĂ« nĂ« github.
Skenari ynĂ« pĂ«r hapjen e konfigurimit â njĂ« program nĂ« gjuhĂ«n Python. TĂ« kaluar nĂ« kodim.
Konfigurojmë konfigurimet për punën e mëtejshme. Do na duhen parametrat e mëposhtëm:
nipyapi.config.nifi_config.host = 'http://nifi:8080/nifi-api' # rruga deri në instancën nifi-api, në të cilën ndërtojmë grupin e proceseve
nipyapi.config.registry_config.host = 'http://nifi-registry:18080/nifi-registry-api' # rruga deri në nifi-registry-api registry
nipyapi.config.registry_name = 'MyBeutifulRegistry' # emri i regjistrit, si do të quhet në instancën nifi
nipyapi.config.bucket_name = 'BucketName' # emri i bucket, nga i cili tërheqim flow
nipyapi.config.flow_name = 'FlowName' # emri i flow, të cilin e tërheqim
Më pas do të filloj të fus emrat e metodave të kësaj biblioteke që janë përshkruar .
Lidhim registrin me instancën e nifi nëpërmjet
nipyapi.versioning.create_registry_clientNë këtë hap mund të shtojmë gjithashtu një kontroll nëse regjistri është shtuar tashmë 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 të flow në kosh
nipyapi.versioning.get_registry_bucketDhe në bazë të bucket të gjetur kërkojmë flow
nipyapi.versioning.get_flow_in_bucketMĂ« pas Ă«shtĂ« e rĂ«ndĂ«sishme tĂ« kuptojmĂ« nĂ«se ky group procesi Ă«shtĂ« shtuar tashmĂ«. Grupi i proceseve vendoset sipas koordinatave dhe mund tĂ« ndodhĂ« qĂ« njĂ« komponent tĂ« mbivendoset me njĂ« tĂ« dytĂ«. E kam kontrolluar, diçka e tillĂ« ndodh đ PĂ«r tĂ« marrĂ« tĂ« gjitha grupet e proceseve tĂ« shtuar, pĂ«rdorim metodĂ«n
nipyapi.canvas.list_all_process_groupsdhe më pas mund të kërkojmë, për shembull, sipas emrit.
Nuk nuk do ta pĂ«rshkruaj procesin e azhurnimit tĂ« templates, do tĂ« them vetĂ«m se nĂ«se nĂ« versionin e ri tĂ« templates shtohen procesorĂ«t, nuk ka probleme me mesazhet nĂ« radhĂ«. MegjithatĂ«, nĂ«se procesorĂ«t hiqen, mund tĂ« shfaqen probleme (nifi nuk lejon heqjen e procesorĂ«ve nĂ«se para tyre ka njĂ« radhĂ« mesazhesh). NĂ«se jeni tĂ« interesuar si e zgjidhja kĂ«tĂ« problem â mĂ« shkruani, tĂ« diskutojmĂ« pĂ«r kĂ«tĂ« çështje. Kontaktet janĂ« nĂ« fund tĂ« artikullit. Le tĂ« kalojmĂ« nĂ« hapin e shtimit tĂ« grupit tĂ« proceseve.
Gjatë depurimit të skriptit, u përballa me një karakteristikë që nuk gjithmonë tërheq versionin më të fundit të flow, ndaj rekomandoj së pari të sqaroni këtë version:
nipyapi.versioning.get_latest_flow_verShkarkojmë grupin e proceseve:
nipyapi.versioning.deploy_flow_versionNisëm procesorët:
nipyapi.canvas.schedule_process_groupNë bllokun në lidhje me CLI është përmendur se në grupin e proceseve të largë nuk përfshihet automatikisht transmetimi i të dhënave? Gjatë zbatimit të skenarit u përballa me këtë problem gjithashtu. Në atë kohë, nuk arrita të aktivizoj transmetimin e të dhënave përmes API, prandaj vendosa të kontaktoj një zhvillues të bibliotekës NiPyAPI dhe të kërkoj këshilla/pomoc. Zhvilluesi më ktheu përgjigje, ne diskutuam problemin dhe ai më tha se i duhej kohë "të kontrollonte disa gjëra". Dhe, pas disa ditësh, erdhi një email, në të cilin ishte përshkruar një funksion në Python, që zgjidhte problemin tim të aktivizimit!!! Në atë kohë, versioni i NiPyAPI ishte 0.13.3 dhe natyrisht, nuk kishte asgjë të tillë. Por në versionin 0.14.0, që doli së fundmi, ky funksion tashmë ishte përfshirë në bibliotekë. E mirëpritur,
nipyapi.canvas.set_remote_process_group_transmissionKështu, me ndihmën e bibliotekës NiPyAPI, lidhem me regjistrin, ngarkoj flow dhe madje aktivizoj procesorët dhe transmetimin e të dhënave. Më pas mund të përsos kodin, të shtoj kontrolle të ndryshme, regjistrim dhe gjithçka tjetër. Por kjo është një histori krejt tjetër.
Nga të gjitha opsionet e automatizimit që kam shqyrtuar, ajo e fundit më duket më funksionale. Së pari, është akoma kod në Python, në të cilin mund të integrohen kode ndihmëse dhe të shfrytëzohen të gjitha përfitimet e gjuhës së programimit. Së dyti, projekti NiPyAPI po zhvillohet aktivisht dhe në rast problemesh mund të kontaktohet zhvilluesi. Së treti, NiPyAPI është një mjet më i përshtatshëm për ndërveprimin me NiFi për zgjidhjen e sfidave më të komplikuara. Për shembull, në përcaktimin nëse radhët e mesazheve janë bosh tani në fluks dhe nëse mund të përditësohet grupi i proceseve.
KĂ«tu pĂ«rfundoj. Kam pĂ«rshkruar 3 qasje pĂ«r automatizimin e shpĂ«rndarjes sĂ« fluksit nĂ« NiFi, pengesat qĂ« mund tĂ« pĂ«rballen zhvilluesit dhe kam sjellĂ« kod funksional pĂ«r automatizimin e shpĂ«rndarjes. NĂ«se ju intereson kjo temĂ« njĂ«soj si muaâ
Burimi: habr.com
