Mindenkinek szia!

A feladat a következő: van egy flow, amely a fenti ábrán látható, és ezt N szerverre kell telepíteni. Apache NiFi. A flow egy teszt, amely során fájl generálódik és elküldésre kerül egy másik NiFi példányba. Az adatok átvitele a NiFi Site to Site protokoll segítségével történik.

A NiFi Site to Site (S2S) egy biztonságos, könnyen konfigurálható mód az adatok átadására NiFi példányok között. Az S2S működéséről itt olvashat. dokumentáció Fontos, hogy ne felejtsük el konfigurálni a NiFi példányt, hogy engedélyezze az S2S-t. További információ. itt.

Amikor S2S-en keresztül történik az adatok átvitele, az egyik példány kliensként, a másik szerverként működik. A kliens küldi az adatokat, a szerver fogadja azokat. Két módja van az adatok átadásának közöttük:

1. Push. Az adatok a kliens példányból a Remote Process Group (RPG) segítségével kerülnek küldésre. A szerver példányban az adatok Input Port-en keresztül érkeznek.
2. Húzás. A szerver RPG-n keresztül fogadja az adatokat, a kliens az Output porton keresztül küldi azokat.

A telepítendő flow-t az Apache Registryben tároljuk.

Az Apache NiFi Registry az Apache NiFi alprojekteként működik, és egy eszközként szolgál a flow-k tárolására és verziókontrolljára. Olyan, mint a GIT. Az instalálásról, konfigurálásról és a registry használatáról információt itt talál. hivatalos dokumentációban. A tárolandó flow-t process group-ba egyesítjük, és ebben a formában tárolódik a registryben. Ebben a cikkben visszatérünk még erre.

Az elején, amikor N kicsi szám, a flow kézi átvitele és frissítése elfogadható időn belül történik.

De N növekedésével egyre több probléma merül fel:

1. az idő, amely a flow frissítéséhez szükséges, nő. Minden szerverre be kell lépni.
2. mint például a sablonok frissítési hibái. Itt frissítve lett, de itt elfelejtettük.
3. az emberi hibák a sok egyforma művelet végrehajtása során.

Mindez oda vezet, hogy automatizálnunk kell a folyamatot. Kipróbáltam a következő megoldásokat a feladat megoldására:

1. MiNiFi használata a NiFi helyett.
2. NiFi CLI.
3. NiPyAPI.

MiNiFi használata.

Apache MiNiFi. — az Apache NiFi alprojektje. A MiNiFy egy kompakt ügynök, amely ugyanazokat a processzorokat használja, mint a NiFi, lehetővé téve a NiFi-vel azonos flow-k létrehozását. Az ügynök könnyűsége részben annak köszönhető, hogy a MiNiFy-nak nincs grafikus felülete a flow konfigurálásához. A grafikus felület hiánya miatt a flow MiNiFy-be történő szállításának problémáját meg kell oldani. Mivel a MiNiFy-t aktívan használják az IoT területén, sok összetevő van, és a flow-k végső példányokhoz történő szállítását automatizálni kell. Ismerős feladat, ugye?

Egy másik alprojekt segíthet a feladat megoldásában — a MiNiFi C2 Server. Ez a termék a konfigurációk telepítési architektúrájának középpontjaként szolgál. Az előkörnyezet konfigurálásáról a cikkben Habrán található információk részletesen leírják a feladat megoldásához szükséges lépéseket. A MiNiFi a C2 serverrel automatikus üzemmódban frissíti a konfigurációt. Ennek az megoldásnak az egyetlen hátránya, hogy sablonokat kell létrehozni a C2 Serveren, a registry-be történő egyszerű elköteleződés nem elegendő.

A fenti cikkben leírt megoldás működőképes, és nem nehéz megvalósítani, de ne felejtsük el a következőket:

1. A MiNiFy-ben nem találhatók meg az összes processzor a NiFi-ból.
2. A MiNiFy processzorai elmaradnak a NiFi processzorainak verzióitól.

A cikk írásakor a legújabb NiFi verzió — 1.9.2. A MiNiFi legújabb verziójának processzora 1.7.0. A processzorokat hozzá lehet adni a MiNiFy-hoz, de a NiFi és MiNiFi processzorai közötti verziókülönbségek miatt ez nem mindig működik.

NiFi CLI.

According to leírása a hivatalos weboldalon, ez az eszköz a NiFi és a NiFi Registry közötti együttműködés automatizálására szolgál a flow-k szállítása vagy a folyamatok kezelésének területén. Az eszköz használatához először le kell tölteni innen.

Indítjuk az alkalmazást

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

Írd be a 'help' parancsot, hogy megtekintsd a rendelkezésre álló parancsok listáját, használd a tabot az automatikus kiegészítéshez.

Ahhoz, hogy betöltsük a szükséges flow-t a registry-ből, tudnunk kell a vödör azonosítóit (bucket identifier) és a flow azonosítóját (flow identifier). Ezeket az adatokat vagy a cli-n, vagy a NiFi registry webes felületén találhatjuk meg. A webes felület így néz ki:

CLI segítségével így működik:

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

A process group importálása a registry-ből:

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

Fontos pillanat — a process group telepítéséhez bármely NiFi példányt megadhatunk hosztként.

A process group has been added with stopped processors; they need to be started.

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

Great, the processors have started. However, according to the task's conditions, we need the NiFi instances to send data to other instances. Let's assume that the chosen method for data transmission to the server is Push. To organize data transmission, we need to enable transmitting on the added Remote Process Group (RPG), which is already included in our flow.

I couldn't find a way to enable data transmission in the CLI documentation and other sources. If you know how to do this, please write in the comments.

Since we have bash and we're ready to go all the way, let's find a solution! We can use the NiFi API to address this issue. We will use the following method; the ID is taken from the examples above (in our case, it's 7f522a13-016e-1000-e504-d5b15587f2f3). Description of NiFi API methods. itt.

In the body, you need to send JSON of the following format:

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

Parameters that need to be filled out for it to work:
state — data transmission status. Available options are TRANSMITTING to enable data transmission, STOPPED to disable it.
version — processor version.

The version will be 0 by default upon creation, but you can obtain these parameters using the method.

For bash script enthusiasts, this method may seem suitable, but it’s a bit challenging for me—bash scripts are not my favorite. The next method seems more interesting and convenient in my opinion.

NiPyAPI.

NiPyAPI is a library for Python to interact with NiFi instances. The documentation page contains the necessary information to work with the library. Quick start is described in. projekten a githubon.

Our script for deploying the configuration is a program in Python. Let's move on to coding.
We are setting up the configurations for further work. We will need the following parameters:

nipyapi.config.nifi_config.host = 'http://nifi:8080/nifi-api' #path to the nifi-api instance where the process group is deployed
nipyapi.config.registry_config.host = 'http://nifi-registry:18080/nifi-registry-api' #path to the nifi-registry-api registry
nipyapi.config.registry_name = 'MyBeautifulRegistry' #registry name, as it will be called in the nifi instance
nipyapi.config.bucket_name = 'BucketName' #bucket name from which we are pulling the flow
nipyapi.config.flow_name = 'FlowName' #flow name that we are pulling

Next, I will list the method names of this library that are described. itt.

Connect the registry to the nifi instance using.

nipyapi.versioning.create_registry_client

Ebben a lépésben hozzáadhatunk egy ellenőrzést is, hogy a registry már hozzá van adva az instanszhoz, ehhez a következő metódust használhatjuk

nipyapi.versioning.list_registry_clients

Keresünk egy bucketet a további flow kereséséhez a kosárban

nipyapi.versioning.get_registry_bucket

A megtalált bucket alapján keresünk flow-t

nipyapi.versioning.get_flow_in_bucket

Fontos megérteni, hogy ez a process group már hozzá lett-e adva. A process group koordináták alapján helyezkedik el, és előfordulhat, hogy egy alkatrészre ráhelyezik a másikat. Ellenőriztem, hogy ez megtörténhet 🙂 Az összes hozzáadott process group megkapásához használjuk a következő metódust

nipyapi.canvas.list_all_process_groups

és ezután például a név alapján is kereshetünk.

Nem fogom részletezni a sablon frissítési folyamatát, csupán annyit mondok, hogy ha az új sablonverzióban processzorok kerülnek hozzáadásra, akkor nincs probléma a sorokban lévő üzenetek meglétével. Azonban ha processzorok törlésre kerülnek, akkor problémák merülhetnek fel (a NiFi nem engedi törölni a processzort, ha előtte sorakozik üzenetek egy sora). Ha érdekel, hogyan oldottam meg ezt a problémát — kérlek, írj nekem, beszéljük meg ezt a kérdést. A kapcsolati adatok a cikk végén találhatók. Lépjünk a process group hozzáadásának lépésére.

A szkript debuggolása közben szembesültem azzal a sajátossággal, hogy nem mindig a legutolsó flow verziót húzza be, ezért ajánlom, hogy először pontosítsuk ezt a verziót:

nipyapi.versioning.get_latest_flow_ver

Deployáljuk a process groupot:

nipyapi.versioning.deploy_flow_version

Elindítjuk a processzorokat:

nipyapi.canvas.schedule_process_group

A CLI blokkban azt írták, hogy a remote process groupban automatikusan nem történik adatátvitel? A szkript megvalósítása során én is találkoztam ezzel a problémával. Abban az időben nem sikerült elindítani az adatátvitelt API segítségével, ezért úgy döntöttem, hogy írok a NiPyAPI könyvtár fejlesztőjének és tanácsot kérdek. A fejlesztő válaszolt, megbeszéltük a problémát, és azt mondta, hogy időre van szüksége, hogy 'ellenőrizzen valamit'. És íme, néhány nappal később érkezik egy levél, amelyben egy Python funkciót írtak, ami megoldja a problémámat a futtatással kapcsolatban!!! Abban az időben a NiPyAPI verziója 0.13.3 volt, és természetesen nem tartalmazott ilyen funkciót. De a legutóbb megjelent 0.14.0 verzióba bekerült ez a funkció. Üdvözöljük,

nipyapi.canvas.set_remote_process_group_transmission

Tehát a NiPyAPI könyvtár segítségével csatlakoztattuk a registry-t, telepítettük a flow-t, sőt a processzorokat is elindítottuk, és adatátvitelt is végeztünk. Most már finomíthatjuk a kódot, különböző ellenőrzéseket, naplózást és egyebeket adhatunk hozzá. De ez már egy teljesen más történet.

A megvizsgált automatizálási lehetőségek közül az utolsó tűnt számomra a legjobban működőnek. Először is, ez egy python kód, amelybe beágyazhatunk segédprogramozási kódot, és élvezhetjük a programozási nyelv minden előnyét. Másodsorban, a NiPyAPI projekt aktívan fejlődik, és ha problémák merülnek fel, írhatunk a fejlesztőnek. Harmadsorban, a NiPyAPI egy rugalmasabb eszköz a NiFi-val való együttműködéshez összetett feladatok megoldásában. Például annak meghatározásában, hogy jelenleg üresek-e a szolgáltatás üzenetsorai, és hogy frissíthetjük-e a process group-ot.

Ezzel véget is értünk. Három megközelítést írtam le a flow automatizált szállítására a NiFi-ba, azokat a buktatókat, amelyekkel a fejlesztő szembesülhet, és bemutattam a működő kódot az automatizálásra. Ha téged is így érdekel ez a téma — írj!

Forrás: habr.com