Apache NIFI – A gyakorlati lehetőségek rövid áttekintése

Bevezetés

Így esett, hogy a jelenlegi munkahelyemen meg kellett ismerkednem ezzel a technológiával. Kezdeném egy kis háttérrel. A következő találkozón azt mondták csapatunknak, hogy létre kell hoznunk az integrációt ismert rendszer. Az integráció alatt azt értettem, hogy ez a jól ismert rendszer HTTP-n keresztül küld nekünk kéréseket egy adott végpontra, mi pedig furcsa módon SOAP üzenet formájában küldjük vissza a válaszokat. Minden egyszerűnek és triviálisnak tűnik. Ebből az következik, hogy szüksége van...

Feladat

Hozzon létre 3 szolgáltatást. Az első ezek közül az adatbázisfrissítési szolgáltatás. Ez a szolgáltatás, amikor új adatok érkeznek egy harmadik féltől származó rendszertől, frissíti az adatbázisban lévő adatokat, és CSV formátumú fájlt generál, hogy átvigye azokat a következő rendszerbe. A második szolgáltatás végpontja az FTP szállítási szolgáltatás, amely fogadja az átvitt fájlt, érvényesíti azt, és FTP-n keresztül fájltárolóba helyezi. A harmadik szolgáltatás, a fogyasztói adatátviteli szolgáltatás az első kettővel aszinkron módon működik. Külső rendszertől kap egy kérést a fent tárgyalt fájl fogadására, átveszi a kész válaszfájlt, módosítja azt (frissíti az azonosítót, leírást, linkToFile mezőket), és SOAP üzenet formájában elküldi a választ. Vagyis az összkép a következő: az első két szolgáltatás csak akkor kezdi meg a munkáját, ha megérkeztek a frissítéshez szükséges adatok. A harmadik szolgáltatás folyamatosan működik, mert sok az információfogyasztó, percenként körülbelül 1000 adatigénylés. A szolgáltatások folyamatosan elérhetők, és példányaik különböző környezetekben találhatók, mint például teszt, demó, előgyártás és prod. Az alábbiakban bemutatjuk, hogyan működnek ezek a szolgáltatások. Azonnal tisztázom, hogy néhány részletet leegyszerűsítettek a szükségtelen bonyolultság elkerülése érdekében.

Műszaki elmélyülés

A probléma megoldásának tervezésekor először úgy döntöttünk, hogy Java nyelven készítünk alkalmazásokat Spring keretrendszer, Nginx balancer, Postgres adatbázis és egyéb technikai és nem annyira technikai dolgok felhasználásával. Mivel a műszaki megoldás kidolgozásának ideje lehetővé tette, hogy más megközelítéseket is mérlegeljünk a probléma megoldására, a tekintetünk a bizonyos körökben divatos Apache NIFI technológiára esett. Azonnal elmondom, hogy ez a technológia lehetővé tette számunkra, hogy észrevegyük ezt a 3 szolgáltatást. Ebben a cikkben egy fájlszállítási szolgáltatás és egy adatátviteli szolgáltatás fejlesztését ismertetem a fogyasztók felé, de ha a cikk hasznos, akkor az adatbázisban lévő adatok frissítésére szolgáló szolgáltatásról is írok.

Mi ez

Az NIFI egy elosztott architektúra az adatok gyors párhuzamos betöltéséhez és feldolgozásához, számos pluginhez a forrásokhoz és átalakításokhoz, a konfigurációk verziózásához és még sok máshoz. Jó bónusz, hogy nagyon könnyen használható. A triviális folyamatok, mint például a getFile, sendHttpRequest és mások négyzetként ábrázolhatók. Minden négyzet egy folyamatot reprezentál, melynek kölcsönhatása az alábbi ábrán látható. Részletesebb dokumentáció készült a folyamatbeállítási interakciókról itt , azoknak, akik beszélnek oroszul - itt. A dokumentáció tökéletesen leírja a NIFI kicsomagolását és futtatását, valamint a folyamatok, más néven négyzetek létrehozását.
A cikkírás gondolata hosszas keresgélés és a kapott információk tudatossá strukturálása után született meg, valamint az a vágy, hogy egy kicsit megkönnyítsük a leendő fejlesztők életét.

Példa

Egy példa arra, hogy a négyzetek hogyan hatnak egymásra. Az általános séma meglehetősen egyszerű: HTTP kérést kapunk (Elméletileg egy fájllal a kérés törzsében. A NIFI képességeinek bemutatására ebben a példában a kérés elindítja a fájl fogadásának folyamatát a helyi fájltárolóról ), majd visszaküldjük azt a választ, hogy a kérés megérkezett, ezzel párhuzamosan az FH-tól egy fájl fogadása, majd az FTP-n keresztül az FH-ba történő áthelyezés folyamata. Érdemes tisztázni, hogy a folyamatok az úgynevezett flowFile-on keresztül lépnek kölcsönhatásba egymással. Ez az alap entitás a NIFI-ben, amely attribútumokat és tartalmat tárol. A tartalom az adatfolyam fájl által képviselt adat. Ez azt jelenti, hogy durván szólva, ha kap egy fájlt az egyik mezőről, és átviszi egy másikba, akkor a tartalom az Ön fájlja lesz.

Amint látja, ez a kép az általános folyamatot mutatja. HandleHttpRequest - kéréseket fogad el, ReplaceText - választörzset generál, HandleHttpResponse - választ küld. FetchFile - egy fájlt fogad egy fájltárolóról, átviszi a négyzet alakú PutSftp-be - FTP-re helyezi a megadott címre. Most többet erről a folyamatról.

Ebben az esetben a kérés mindennek a kezdete. Nézzük a konfigurációs paramétereit.

Itt minden meglehetősen triviális, kivéve a StandardHttpContextMap - ez egy olyan szolgáltatás, amely lehetővé teszi kérések küldését és fogadását. Részletesebben és akár példákkal is láthatja - itt

Ezután nézzük meg a négyzet ReplaceText konfigurációs paramétereit. Érdemes odafigyelni a ReplacementValue-ra – ez az, amit válasz formájában visszakap a felhasználó. A beállításokban módosíthatja a naplózás mértékét, láthatja a naplókat {ahol kicsomagolta a nifit}/nifi-1.9.2/logs, vannak hiba/siker paraméterek is - ezek alapján a folyamat egészét szabályozhatja . Azaz sikeres szövegfeldolgozás esetén a felhasználónak való válaszküldés folyamata lesz meghívva, egy másik esetben pedig egyszerűen naplózzuk a sikertelen folyamatot.

A HandleHttpResponse tulajdonságokban nincs semmi különösebben érdekes, kivéve a válasz sikeres létrehozásának állapotát.

A kérést és a választ rendeztük – térjünk át a fájl fogadására és az FTP szerverre való elhelyezésére. FetchFile - fogad egy fájlt a beállításokban megadott elérési úton, és továbbítja a következő folyamatnak.

Ezután a PutSftp négyzet - a fájlt a fájltárolóba helyezi. Alább láthatjuk a konfigurációs paramétereket.

Érdemes odafigyelni arra, hogy minden négyzet külön folyamat, amit el kell indítani. Megnéztük a legegyszerűbb példát, amely nem igényel bonyolult testreszabást. Ezután egy kicsit bonyolultabban nézzük meg a folyamatot, ahol egy kicsit írunk a barázdákról.

Bonyolultabb példa

A SOAP üzenet módosítási folyamata miatt kicsit bonyolultabbnak bizonyult az adatátviteli szolgáltatás a fogyasztónak. Az általános folyamat az alábbi ábrán látható.

Itt sem különösebben bonyolult az ötlet: kaptunk egy kérést a fogyasztótól, hogy adatokra van szüksége, küldtünk egy választ, hogy üzenetet kapott, elindítottuk a válaszfájl fogadásának folyamatát, majd bizonyos logikával szerkesztettük, majd átadta a fájlt a fogyasztónak SOAP üzenet formájában a szervernek.

Azt hiszem, nem szükséges újra leírni azokat a négyzeteket, amelyeket fent láttunk - térjünk át az újakra. Ha bármilyen fájlt szerkeszteni kell, és a szokásos ReplaceText típusú négyzetek nem megfelelőek, akkor saját szkriptet kell írnia. Ezt az ExecuteGroogyScript négyzet segítségével lehet megtenni. Beállításait az alábbiakban mutatjuk be.

Két lehetőség van a szkript betöltésére ebbe a négyzetbe. Az első egy fájl letöltése egy szkripttel. A második egy szkript beszúrása a scriptBody-ba. Amennyire én tudom, az executeScript négyzet több nyelvet támogat - az egyik a groovy. Csalódást fogok okozni a java fejlesztőknek - ilyen négyzetekben nem lehet Java-ban szkripteket írni. Azok számára, akik igazán szeretnének, létre kell hozniuk saját egyéni négyzetet, és hozzá kell adniuk a NIFI rendszerhez. Ezt az egész műveletet egy elég hosszú, tamburával való tánc kíséri, amivel ebben a cikkben nem foglalkozunk. A groovy nyelvet választottam. Az alábbiakban egy tesztszkript látható, amely egyszerűen fokozatosan frissíti az azonosítót egy SOAP-üzenetben. Fontos megjegyezni. Kiveszed a fájlt a flowFile-ból és frissíted, ne felejtsd el, hogy frissítve vissza kell tenned oda. Azt is érdemes megjegyezni, hogy nem minden könyvtár szerepel benne. Előfordulhat, hogy mégis importálnia kell valamelyik lib-et. Egy másik hátránya, hogy a forgatókönyvet ezen a téren meglehetősen nehéz hibakeresni. Van mód a NIFI JVM-hez való csatlakozásra és a hibakeresési folyamat elindítására. Személy szerint elindítottam egy helyi alkalmazást, és szimuláltam egy fájl fogadását a munkamenetből. Helyben végeztem hibakeresést is. A szkript betöltésekor megjelenő hibákat meglehetősen könnyű a Google számára, és a NIFI maga írja a naplóba.

import org.apache.commons.io.IOUtils
import groovy.xml.XmlUtil
import java.nio.charset.*
import groovy.xml.StreamingMarkupBuilder

def flowFile = session.get()
if (!flowFile) return
try {
    flowFile = session.write(flowFile, { inputStream, outputStream ->
        String result = IOUtils.toString(inputStream, "UTF-8");
        def recordIn = new XmlSlurper().parseText(result)
        def element = recordIn.depthFirst().find {
            it.name() == 'id'
        }

        def newId = Integer.parseInt(element.toString()) + 1
        def recordOut = new XmlSlurper().parseText(result)
        recordOut.Body.ClientMessage.RequestMessage.RequestContent.content.MessagePrimaryContent.ResponseBody.id = newId

        def res = new StreamingMarkupBuilder().bind { mkp.yield recordOut }.toString()
        outputStream.write(res.getBytes(StandardCharsets.UTF_8))
} as StreamCallback)
     session.transfer(flowFile, REL_SUCCESS)
}
catch(Exception e) {
    log.error("Error during processing of validate.groovy", e)
    session.transfer(flowFile, REL_FAILURE)
}

Valójában itt ér véget a négyzet testreszabása. Ezután a frissített fájl átkerül a négyzetbe, amely felelős a fájl kiszolgálóra való elküldéséért. Alább láthatók a négyzet beállításai.

Leírjuk a SOAP üzenet továbbításának módját. Megírjuk, hogy hol. Ezután jeleznie kell, hogy ez a SZAPPAN.

Adjon hozzá több tulajdonságot, például gazdagépet és műveletet (soapAction). Mentjük és ellenőrizzük. További részleteket láthat a SOAP-kérések küldésével kapcsolatban itt

Több lehetőséget is megvizsgáltunk az NIFI folyamatok használatára. Hogyan lépnek kapcsolatba egymással, és mi a valódi hasznuk? A figyelembe vett példák teszt jellegűek, és kissé eltérnek attól, ami a harcban valójában történik. Remélem, ez a cikk egy kicsit hasznos lesz a fejlesztők számára. Köszönöm a figyelmet. Ha kérdésed van írj. Megpróbálok válaszolni.

Forrás: will.com