Apache NIFI — Lühike ülevaade võimalustest praktikas

Sissejuhatus

Minu praeguses töökohas olen pidanud tutvuma selle tehnoloogiaga. Alustan väikese taustaga. Ühes järgmisel kohtumisel öeldi meie meeskonnale, et peame looma integreerimise tuntud süsteemiga. Integreerimise all mõistetakse, et see tuntud süsteem saadab meile HTTP kaudu päringuid kindlale lõpp-punktile, ja meie, nagu see ka poleks imelik, saadame tagasi vastuseid SOAP sõnumite kujul. Tundub, et kõik on lihtne ja ilmselge. Selle tulemusena tuleb...

Ülesanne

Luuakse 3 teenust. Esimene neist on Andmebaasi uuendamise teenus. See teenus, kui saabub uusi andmeid kolmandalt süsteemilt, uuendab andmeid andmebaasis ja genereerib mingi CSV formaadis faili, et edastada see järgmisele süsteemile. Kutsutakse esile teise teenuse, Faili transportimise teenuse, lõpp-punkt, mis saab saadetud faili, valideerib selle ja laadib FTP kaudu failide säilitamisesse. Kolmas teenus, Andmete edastamise teenus tarbijale, töötab asünkroonselt kahe esimese teenusega. See võtab vastu päringu väliselt süsteemilt, et saada varem mainitud faili, võtab valmis vastuse faili, muudab seda (uuendab id, description, linkToFile väljad) ja edastab vastuse SOAP sõnumina. Kokkuvõttes on olukord järgmine: esimesed kaks teenust alustavad oma tööd ainult siis, kui saabuvad andmed uuendamiseks. Kolmas teenus töötab pidevalt, kuna teabe tarbijaid on palju, ligikaudu 1000 päringut andmete saamiseks minutis. Teenused on pidevalt kättesaadavad ja nende instantsid asuvad erinevates keskkondades, nagu test, demo, ettevalmistus ja tootmine. Allpool on toodud nende teenuste töö skeem. Märgin kohe, et mõned detailid on lihtsustatud, et vältida liigset keerukust.

Tehniline süvenemine

Probleemi lahendamiseks otsustasime esmalt välja arendada rakendusi Java keeles, kasutades Spring Framework'i, Nginx'i koormuse tasakaalustajana, Postgres'i andmebaasi ning muid tehnilisi ja vähem tehnilisi lahendusi. Kuna tehnilise lahenduse süvendamine võimaldas arvesse võtta ka teisi lähenemisviise, suunati pilk uuenduslikule Apache NIFI tehnoloogiale, mis on üha populaarsem teatud ringkondades. Pean kohe ütlema, et see tehnoloogia aitas meil avastada need 3 teenust. Selles artiklis käsitletakse faili transportimise teenuse ja andmete edastamise teenuse väljatöötamist, kuid kui artikkel osutub huvitavaks, kirjutan ka andmete värskendamise teenusest andmebaasis.

Mis see on

NIFI esindab enda all jaotatud arhitektuuri kiireks paralleelseks andmete laadimiseks ja töötlemiseks, seal on suur hulk pistikprogramme allikatele ja teisendustele, konfigureerimise versioonimine ja palju muud. Meeldiv boonus on see, et see on väga lihtne kasutada. Triviaalset protsessi, näiteks getFile, sendHttpRequest ja teised, saab kujutada ruutudena. Iga ruut esindab mingit protsessi, mille vahelisi seoseid saab näha alloleval joonisel. Üksikasjalikum dokumentatsioon protsesside seadistamisest ja koostamisest on kirjutatud siin , nende jaoks, kes räägivad vene keelt — siin. Dokumentatsioonis on suurepäraselt selgitatud, kuidas NIFI lahti pakkida ja käivitada ning samuti, kuidas luua protsesse, mida nimetatakse ruutudeks.
Artikli kirjutamise idee tekkis pärast pikaajalisi otsinguid ja saadud teabe struktureerimist millekski mõistetavaks, samuti soovist natuke kergendada tulevaste arendajate elu.

Näide

Vaatame, kuidas ruudud omavahel suhtlevad. Üldine skeem on üsna lihtne: saame HTTP päringu (teoreetiliselt failiga päringu kehas. NIFI võimaluste demonstreerimiseks algab käesolevas näites päring faili saamise protsessi kohalikust failihaldusest), seejärel saadame vastuse tagasi, et päring on saadud, samal ajal käivitub faili saamise protsess failihalduses ja seejärel tuleb faili edastamine FTP kaudu. On oluline märkida, et protsessid suhtlevad omavahel nn flowFile'i kaudu. See on NIFI põhielement, mis salvestab atribuute ja sisu. Sisu tähendab andmeid, mis esindavad voofaili. Teisisõnu, kui saad faili ühest ruudust ja edastad selle teise, on sisu see fail.

Nagu näete — sellel joonisel on kujutatud üldine protsess. HandleHttpRequest — töötab päringute vastuvõtjana, ReplaceText — genereerib vastuse keha, HandleHttpResponse — saadab vastuse. FetchFile — saadab faili failihaldusest ja edastab selle ruudule PutSftp — asetab selle faili FTP-sse, määratud aadressile. Nüüd vaatame seda protsessi põhjalikumalt.

Antud juhul on kõik algus päringust. Vaatame selle konfiguratsiooniparameetreid.

Siin on kõik üsna triviaalne, välja arvatud StandartHttpContextMap — see on teenus, mis võimaldab päringute saatmist ja vastuvõttu. Täiendavaid üksikasju ja näiteid saab vaadata — siin

Järgmisena vaatame ReplaceText ruudu konfiguratsiooniparameetreid. Siin tasub tähelepanu pöörata ReplacementValue'le — see on see, mis naaseb kasutajale vastusena. Seadetest saab reguleerida logimise taset, logisid saab vaadata {kuhu pakkisite nifi}/nifi-1.9.2/logs seal on ka failure/success parameetrid — nende põhjal saad reguleerida protsessi üldiselt. St juhul, kui teksti säilitamine on õnnestunud — käivitatakse protsess kasutajale vastuse saatmiseks, vastasel juhul lihtsalt logime ebaõnnestunud protsessi.

HandleHttpResponse omadustes pole erilist huvitavat, välja arvatud staatus, kui vastus on edukalt loodud.

Oleme aru saanud päringu ja vastuse osast — liikume edasi faili saamise ja selle paigaldamise juurde FTP-serverisse. FetchFile — saadab faili määratud tee järgi ja edastab selle järgmisele protsessile.

Ja järgmine ruut PutSftp - paigutab faili failihoidlasse. Konfiguratsiooni määrangud on allpool nähtavad.

Tasub pöörata tähelepanu sellele, et iga ruut on eraldi protsess, mis peab olema käivitatud. Vaatasime üle kõige lihtsama näite, mis ei vaja mingit keerulist kohandamist. Järgmine on veidi keerulisem protsess, kus kirjutame natuke groovi.

Täpsem näide

Andmete edastamise teenus osutus veidi keerulisemaks kuna see hõlmas SOAP-sõnumi muutmise protsessi. Üldine protsess on kujutatud alloleval joonisel.

Siin pole idee eriti keeruline: saime tarbijalt päringu, et tal on andmeid vaja, saatsime vastuse, et oleme sõnumi kätte saanud, käivitasime faili vastuse saamise protsessi, seejärel redigeerisime seda teatud loogika kohaselt, pärast mida edastatud fail tarbijale SOAP-sõnumina serverisse.

Ma arvan, et ei ole vaja uuesti kirjeldada neid ruute, mida me nägime eespool - liigume kohe uute juurde. Kui teil on vaja redigeerida mõnda faili ja tavalised ruudud nagu ReplaceText ei sobi, peate kirjutama enda skripti. Seda saab teha ExecuteGroogyScript ruudu abil. Selle seadistused on allpool esitatud.

On this square, there are two options for uploading a script. The first is by uploading a file with the script. The second is by inserting the script into scriptBody. As far as I know, the executeScript square supports multiple programming languages — one of them being Groovy. I must disappoint Java developers — scripts cannot be written in such squares using Java. For those who are very eager — you need to create your custom square and integrate it into the NIFI system. This entire operation is accompanied by quite lengthy sacrifices, which we won't cover in this article. I chose the Groovy language. Below is a test script that simply increments the ID in the SOAP message. It is important to note that you take the file from flowFile, update it, and do not forget to place it back there, updated. Also, it is worth noting that not all libraries are connected. It may turn out that you have to import one of the libs. Another downside is that debugging the script in this square is quite difficult. There is a way to connect to the NIFI JVM and start the debugging process. Personally, I ran a local application and mimicked receiving a file from a session. I also debugged locally. The errors that pop up during the script upload can be easily googled and are logged by NIFI itself.

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)
}

This is where the customization of the square ends. The updated file is then sent to the square that handles sending the file to the server. Below are the settings for this square.

We describe the method that will be used to send the SOAP message. We specify where. Next, we need to indicate that this is specifically SOAP.

Lisame mõned omadused, nagu host ja tegevus(soapAction). Salvestame ja kontrollime. Soodsamate SOAP-päringute saatmise kohta saab lähemalt lugeda. siit

Oleme uurinud mitmeid viise NIFI protsesside kasutamiseks. Kuidas need omavahel suhtlevad ja milline on nende tegelik kasu. Uuritud näited on katse olemuselt ja veidi erinevad sellest, mis tegelikult tootmises toimub. Loodan, et see artikkel on arendajatele mõnevõrra kasulik. Aitäh tähelepanu eest. Kui teil on küsimusi, kirjutage julgelt. Proovin vastata.

Allikas: habr.com