Visiem sveiki!

Uzdevums ir šāds — ir plūsma, kas attēlota iepriekšējā attēlā, kuru nepieciešams izvērst uz N serveriem ar Apache NiFi. Plūsma ir testējoša — notiek faila ģenerēšana un tā sūtīšana uz citu NiFi instanci. Datu pārraide notiek, izmantojot NiFi Site to Site protokolu.

NiFi Site to Site (S2S) — drošs, viegli konfigurējams veids datu pārsūtīšanai starp NiFi instancēm. Kā darbojas S2S skatiet dokumentācijā un ir svarīgi neaizmirst konfigurēt NiFi instanci, lai atļautu S2S, skatiet šeit.

Gadījumos, kad runa ir par datu pārsūtīšanu izmantojot S2S — viena instance tiek dēvēta par klientu, otra par serveri. Klients sūta datus, servers tos pieņem. Ir divi veidi, kā konfigurēt datu pārsūtīšanu starp tiem:

1. Push. No klienta instancēm dati tiek sūtīti, izmantojot Remote Process Group (RPG). Servera instancēs dati tiek pieņemti, izmantojot Input Port
2. Pull. Serveris pieņem datus, izmantojot RPG, klients nosūta, izmantojot Output port.

Plūsmu, ko jāpaveic, mēs uzglabājam Apache Registry.

Apache NiFi Registry — apakšprojekts Apache NiFi, kas ir rīks plūsmu glabāšanai un versiju pārvaldībai. Šis ir kaut kas līdzīgs GIT. Informāciju par uzstādīšanu, konfigurāciju un darbu ar registry var atrast oficiālajā dokumentācijā. Plūsmas glabājas process grupā un šādā veidā tiek uzglabāta registry. Par to vēl runāsim raksta turpinājumā.

Sākumā, kad N ir mazs skaitlis, plūsma tiek konsultēta un aktualizēta roku darbs pieņemamā laikā.

Bet pieaugot N, problēmas kļūst arvien vairāk:

1. plūsmas aktualizēšanai nepieciešams vairāk laika. Jāieiet visos servidores
2. notiek šablonu aktualizēšanas kļūdas. Šeit tika atjaunināts, bet šeit aizmirsts
3. cilvēka kļūdas, veicot lielu skaitu vienveidīgu operāciju

Viss šis noved mūs pie secinājuma, ka jāpāriet uz automatizētu procesu. Esmu izmēģinājis šādus risinājumus šai problēmai:

1. Izmanto MiNiFi aizvietojot NiFi
2. NiFi CLI
3. NiPyAPI

MiNiFi izmantošana

Apache MiNiFi — apakšprojekts Apache NiFi. MiNiFy — kompakts aģents, kas izmanto tos pašus procesorus kā NiFi, ļaujot izveidot tādus pašus flow kā NiFi. Aģenta vieglums tiek panākts, arī neizveidojot grafisko interfeisu flow konfigurācijai. Grafiskā interfeisa trūkums MiNiFy nozīmē, ka ir jārisina flow piegādes problēma uz minifi. Ņemot vērā, ka MiNiFy ir aktīvi izmantots IOT, komponentu ir daudz, un procesam, lai piegādātu flow uz gala eksemplāriem minifi, jābūt automatizētam. Pazīstama uzdevums, vai ne?

Šādu uzdevumu var palīdzēt atrisināt vēl viens apakšprojekts — MiNiFi C2 Server. Šis produkts ir paredzēts, lai būtu centrālā punkta arhitektūrā konfigurāciju izdalīšanai. Kā konfigurēt vidi — ir aprakstīts šajā rakstā Habrā un ir pietiekami informācijas, lai atrisinātu noteikto uzdevumu. MiNiFi kopā ar C2 serveri automātiskā režīmā atjaunina konfigurāciju pie sevis. Vienīgais šā piegājiena trūkums — ir jāizveido veidnes C2 Server, vienkāršs reģistrs komits nav pietiekams.

Iepriekš rakstītajā rakstā aprakstītais variants ir efektīvs un vienkārši īstenojams, taču jāatceras sekojošais:

1. Minifi nav visu procesoru no nifi
2. Procesoru versijas Minifi atpaliek no procesoru versijām NiFi.

Uz šī publikācijas brīdi pēdējā versija NiFi ir — 1.9.2. Pēdējās versijas MiNiFi procesoru versija ir — 1.7.0. Procesorus var pievienot MiNiFi, taču sakarā ar versiju atšķirībām starp NiFi un MiNiFi procesoriem tas var nedarboties.

NiFi CLI

Pēc aprakstā rīka oficiālajā mājaslapā, šis rīks ir paredzēts automātiskai NiFi un NiFi Registry mijiedarbības nodrošināšanai flow piegādes vai procesu pārvaldīšanas jomā. Lai uzsāktu darbu, šo rīku nepieciešams lejupielādēt no šejienes.

Palaidiet utilītu

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

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

Lai mēs iegūtu nepieciešamo flow no reģistra, mums jāzina groza identifikatori (bucket identifier) un paša flow (flow identifier). Šos datus var iegūt gan caur cli, gan NiFi reģistra tīmekļa saskarnē. Tīmekļa saskarnē tas izskatās šādi:

Izmantojot CLI, tas tiek darīts šādi:

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

Palaidiet importēt process grupu no reģistra:

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

Svarīgs punkts — kā hostu, uz kuru mēs tiekam uz process grupu, var norādīt jebkuru nifi instanci.

Process group pievienots ar apturētiem procesoriem, tos jārestartē.

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

Lieliski, procesori ir uzsākti. Tomēr, saskaņā ar uzdevuma nosacījumiem, mums ir jānodod dati uz citiem NiFi instancēm. Pieņemsim, ka datu nosūtīšanai uz serveri ir izvēlēts Push veids. Lai nodrošinātu datu pārsūtīšanu, jāieslēdz datu pārsūtīšana pie pievienotā Remote Process Group (RPG), kas jau ir iekļauts mūsu plūsmā.

Documetācijā CLI un citos avotos neatradu veidu, kā ieslēgt datu pārsūtīšanu. Ja jūs zināt, kā to izdarīt — lūdzu rakstiet komentāros.

Tā kā mums ir bash un esam gatavi doties līdz galam — atradīsim risinājumu! Var izmantot NiFi API, lai risinātu šo problēmu. Izmantojam šādu metodi, ID ņemot no iepriekš minētajiem piemēriem (mūsu gadījumā tas ir 7f522a13-016e-1000-e504-d5b15587f2f3). NiFi API metožu apraksts. šeit.

Body jānosūta JSON, ar sekojošu struktūru:

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

Parametri, kurus jāaizpilda, lai "nostrādātu":
state — datu pārsūtīšanas statuss. Pieejams TRANSMITTING pār datu pārsūtīšanas ieslēgšanu, STOPPED pār izslēgšanu.
version — procesora versija.

Versija pēc noklusējuma būs 0, kad izveidojat, bet šos parametrus var iegūt, izmantojot metodi.

Bash skriptu cienītājiem šī metode var šķist noderīga, taču man ir grūti — bash skripti nav mans iecienītākais. Nākamā metode, manuprāt, ir interesantāka un ērtāka.

NiPyAPI

NiPyAPI — Python bibliotēka, lai mijiedarbotos ar NiFi instancēm. Dokumentācijas lapa satur nepieciešamo informāciju, lai strādātu ar bibliotēku. Ātrā sākšana aprakstīta. projektā GitHub.

Mūsu skripts konfigurācijas izvietošanai — programma Python valodā. Pārejam pie kodēšanas.
Iestatām konfigurācijas turpmākai darbībai. Mums būs nepieciešami sekojošie parametri:

nipyapi.config.nifi_config.host = 'http://nifi:8080/nifi-api' #ceļš uz nifi-api instanci, kur izvirzām process group
nipyapi.config.registry_config.host = 'http://nifi-registry:18080/nifi-registry-api' #ceļš uz nifi-registry-api registry
nipyapi.config.registry_name = 'MyBeautifulRegistry' #registry nosaukums, kā tas tiks nosaukts nifi instancē
nipyapi.config.bucket_name = 'BucketName' #bucket nosaukums, no kura ievedam plūsmu
nipyapi.config.flow_name = 'FlowName' #plūsmas nosaukums, kuru ievedam

Tālāk esmu ieplānojis iekļaut šīs bibliotēkas metožu nosaukumus, kas aprakstīti. šeit.

Savienojam registry ar nifi instanci, izmantojot.

nipyapi.versioning.create_registry_client

Šajā solī var vēl pievienot pārbaudi, vai reģistrs jau ir pievienots instancēm, izmantojot metodi

nipyapi.versioning.list_registry_clients

Atrast glabātuvi turpmākai flow meklēšanai tvertnē

nipyapi.versioning.get_registry_bucket

Pēc atrastās glabātuves meklējam flow

nipyapi.versioning.get_flow_in_bucket

Tālāk ir svarīgi saprast, vai šī process grupa jau nav pievienota. Process grupa tiek novietota pēc koordinātēm un var rasties situācija, kad viens komponents tiek pārklāts ar otro. Es pārbaudīju, tas var notikt 🙂 Lai iegūtu visas pievienotās process grupas, izmantojam metodi

nipyapi.canvas.list_all_process_groups

un tālāk varam meklēt, piemēram, pēc nosaukuma.

Es neatklāšu šablona atjaunināšanas procesu, tikai teikšu, ka, ja jaunajā šablona versijā tiek pievienoti procesori, tad problēmu ar ziņojumiem rindā nav. Taču, ja procesori tiek dzēsti, tad var rasties problēmas (nifi neļauj dzēst procesoru, ja pirms tā ir uzkrāta ziņojumu rinda). Ja jūs interesē, kā es atrisināju šo problēmu — lūdzu, uzrakstiet man, izrunāsim šo jautājumu. Kontakti raksta beigās. Pārmetīsim pie process grupas pievienošanas soļa.

Taisot skriptu, es saskāros ar īpatnību, ka ne vienmēr tiek ielādēta jaunākā flow versija, tāpēc iesaku vispirms noskaidrot šo versiju:

nipyapi.versioning.get_latest_flow_ver

Veicam process grupas izvietošanu:

nipyapi.versioning.deploy_flow_version

Palaidam procesorus:

nipyapi.canvas.schedule_process_group

CLI blokā tika minēts, ka attālinātajā process grupā automātiski netiek iekļauta datu pārsūtīšana? Realizējot skriptu, es saskāros ar šo problēmu arī. Tajā laikā man neveiksmējās palaist datu pārsūtīšanu, izmantojot API, tāpēc nolēmu uzrakstīt NiPyAPI bibliotēkas izstrādātājam un lūgt padomu/palīdzību. Izstrādātājs man atbildēja, mēs apspriedām problēmu, un viņš rakstīja, ka viņam nepieciešams laiks

nipyapi.canvas.set_remote_process_group_transmission

Tātad ar NiPyAPI bibliotēku mēs pievienojām reģistru, uzstādījām plūsmu un pat palaidām procesorus un datu pārsūtīšanu. Tagad varam sakārtot kodu, pievienot dažādas pārbaudes, žurnālu un visu to. Bet tas jau ir pavisam cita stāsta daļa.

No manā apskatītā automatizācijas variantu klāsta pēdējais šķita visefektīvākais. Pirmkārt, tas joprojām ir Python kods, kuru var papildināt ar palīgtehniskām programmām un izmantot visas programmēšanas valodas priekšrocības. Otrkārt, projekts NiPyAPI aktīvi attīstās, un problēmu gadījumā var sazināties ar izstrādātāju. Treškārt, NiPyAPI tomēr ir elastīgāks instruments mijiedarbībai ar NiFi sarežģītu uzdevumu risināšanai. Piemēram, lai noteiktu, vai ziņojumu rindas patiešām ir tukšas plūsmā un vai var atjaunināt procesu grupu.

Uz šo visu ir viss. Es aprakstīju 3 pieejas plūsmu piegādes automatizācijai NiFi, slēptās problēmas, ar kurām var saskarties izstrādātājs, un sniedzu strādājošu kodu plūsmu piegādes automatizācijai. Ja jūs, tāpat kā mani, interesē šī tēma — rakstiet!

Avots: habr.com