Apache NIFI — Praktiline ülevaade võimalustest

Sissejuhatus

Kuid olen oma praeguses töökohas pidanud selle tehnoloogiaga tutvuma. Alustan väikese taustaga. tuntud süsteemi. All integraatsiooni all mõeldi, et see tuntud süsteem saadab meile HTTP kaudu päringud kindlale lõpp-punktile, ja meie, nagu see ei tundu kummaline, saadame tagasi vastuseid SOAP-sõnumina. Tundub, et kõik on lihtne ja triviaalne. Sellest järeldub, et on vajalik...

Ülesanne

Loo 3 teenust. Esimene neist on Andmebaasi uuendamise teenus. See teenus uuendab andmeid andmebaasis ja genereerib CSV formaadis faili, kui saavad uued andmed välistest süsteemidest, et edastada see järgmisesse süsteemi. Kõneleja on teise teenuse — FTP edastamise teenuse lõpp-punkt, mis saab edastatud faili, valideerib selle ja salvestab FTP kaudu failide salvestusse. Kolmas teenus — Andmete edastamise teenus tarbijale, töötab asünkroonselt kahe esimese teenusega. See võtab vastu päringu välistest süsteemidest, et saada ülal mainitud faili ning võtab valmis vastusfaili, muudab seda (uuendab id, description, linkToFile väljad) ja saadab vastuse SOAP sõnumina. Seega on olukord järgmine: esimesed kaks teenust alustavad oma tööd ainult siis, kui uuendamiseks on andmed saabunud. Kolmas teenus töötab pidevalt, kuna informatsiooni tarbijaid on palju, umbes 1000 päringut andmete saamiseks minutis. Teenused on pidevalt saadaval ja nende instantsid asuvad erinevates keskkondades, nagu test, demo, eeltootmine ja tootmine. Allpool on esitatud nende teenuste töö schema. Selgitan kohe, et mõned detailid on lihtsustatud, et vältida liigset keerukust.

Tehniline süvenemine

Probleemi lahendamise planeerimisel otsustasime esmalt luua Java rakendusi, kasutades Springi raamistikku, Nginx'i koormuse tasakaalustajat, PostgreSQL andmebaasi ja muid tehnilisi ja vähem tehnilisi asju. Kuna aega tehnilise lahenduse väljatöötamiseks oli, võimaldas see kaaluda ka teisi lähenemisviise, ja meie tähelepanu köitis teatud ringkondades populaarne tehnoloogia Apache NIFI. Tõeliselt, see tehnoloogia aitas meil märgata neid kolme teenust. Selles artiklis kirjeldatakse faili edastamise teenuse ja andmete edastamise teenuse arendamist, kuid kui artikkel meeldib, kirjutan ka andmete uuendamise teenusest andmebaasis.

Mis see on

NIFI on jaotatud arhitektuur kiireks paralleelseks andmete laadimiseks ja töötlemiseks, suure hulgapluginatega allikatelt ja teisendustelt, konfigureerimise versioonimine ja palju muud. Tore boonus on see, et seda on väga lihtne kasutada. Lihtsaid protsesse, nagu getFile, sendHttpRequest ja teised, saab kujutada ruutudena. Iga ruut esindab mingit protsessi, mille interaktsiooni saab näha alloleval joonisel. Täiendav dokumentatsioon protsesside seadistamise kohta on kirjutatud siit , venekeelsetele kasutajatele — siit. Dokumentatsioonis on suurepäraselt kirjeldatud, kuidas NIFI lahti pakkida ja käivitada, samuti kuidas luua protsesse, need ongi ruudud
Artikli kirjutamise idee tuli pärast pikaajalist teabe otsimist ja struktureerimist, et muuta see millegiks arusaadavaks, samuti soovist veidi kergendada tulevaste arendajate elu.

Näide

Kuidas ruudud omavahel suhtlevad, on üksikasjalikult illustreeritud järgmistest etappidest. Üldine skeem on üsna lihtne: saadame HTTP päringu (teoorias koos failiga päringu õppusena. NIFI võimaluste demonstreerimiseks algab selles näites päring protsessi, mis saadab faili kohalikust failihaldusest), seejärel saadame tagasi vastuse, et päring on saadud, samal ajal kui algab faili hankimise protsess failihaldusest ja seejärel faili edastamine FTP kaudu teise failihalduse. Oluline on märkida, et protsessid suhtlevad üksteisega tänu nn flowFile'ile. See on NIFI põhiolukord, mis salvestab atribuute ja sisu. Sisu tähendab andmeid, mis esindavad voolufaili. Teisisõnu, kui saate faili ühest ruudust ja edastate selle teise, on sisuks teie fail.

Nagu võite märgata, on sellel joonisel kujutatud üldine protsess. HandleHttpRequest — võtab vastu päringud, ReplaceText — genereerib vastuse keha, HandleHttpResponse — edastab vastuse. FetchFile — saab faili failihaldusest ja edastab selle ruudule PutSftp — salvestab selle faili FTP-sse, määratud aadressile. Nüüd räägime sellest protsessist lähemalt.

Antud juhul on request kõikide algus. Vaatame selle konfiguratsiooniparameetreid.

Siin on kõik üsna banaalne, välja arvatud StandartHttpContextMap — see on teenus, mis võimaldab saata ja vastu võtta päringuid. Üksikasjalikumalt ja isegi näidetega saab vaadata — siit

Järgmisena vaatame ReplaceText ruudu konfiguratsiooniparameetreid. Siin on tähelepanu pöörata ReplacementValue'le — see on see, mis kasutajale vastusena tagasi tuleb. Seadetest saab reguleerida logimise taset, logisid saab vaadata {kuhu on Nifi lahti pakitud}/nifi-1.9.2/logs, seal on ka failure/success parameetrid — nende parameetrite põhjal saab kogu protsessi reguleerida. St, kui teksti töötlemine õnnestub — kutsutakse välja vastuse saatmise protsess, ja muul juhul logime lihtsalt ebaõnnestunud protsessi.

HandleHttpResponse omadustes pole erilist huvitavat, välja arvatud staatuse olemasolu, kui vastuse loomine õnnestub.

Oleme vastuse küsimisega selgeks saanud — liigume edasi faili saamise ja selle paigutamise juurde FTP-serverisse. FetchFile — saab faili määratud seadistustes olevast teest ja edastab selle järgmisele protsessile.

Ja PutSftp-ruut paigutab faili failihoidlasse. Konfiguratsiooni seaded on allpool.

Oluline on märkida, et iga ruut on eraldi protsess, mis tuleb käivitada. Oleme vaadanud kõige lihtsamat näidet, mis ei nõua mingit keerulist kohandamist. Järgmisena vaatame veidi keerulisemat protsessi, kus kirjutame natuke gnuud.

Keerulisem näide

Andmete edastus teenus osutus veidi keerulisemaks SOAP sõnumi muutmisprotsessi tõttu. Üldine protsess on näidatud alloleval joonisel.

Siin on idee samuti mitte eriti keeruline: saime tarbijalt päringu, et ta vajab andmeid, saatsime vastuse, et oleme sõnumi saanud, käivitasime vastuse faili saamise protsessi, seejärel redigeerisime seda teatud loogika järgi ja edastasime faili tarbijale SOAP sõnumina serverisse.

Arvan, et pole mõtet uuesti kirjeldada neid ruute, mida me varem nägime — liigume otse uute juurde. Kui peate redigeerima mõnda faili ja tavalised ruudud, nagu ReplaceText, ei sobi, peate kirjutama oma skripti. Selle saab teha ruudu ExecuteGroogyScript abil. Selle seadistused on toodud allpool.

On kaks varianti skripti laadimiseks selle kuubiku sisse. Esimene on failiga, kus skript on laaditud. Teine — skripti kopeerimine scriptBody sisse. Teatavasti toetab kuubik executeScript mitut programmeerimiskeelt — üks neist on groovy. Java arendajatele pettumuseks — java keeles ei saa kirjutada skripte sellistes kuubikutes. Neile, kes tahavad väga, tuleb luua oma eritellimusel kuubik ja see NIFI süsteemi lisada. Kogu see operatsioon kaasneb üsna pika tantsuga trikolo, millega me selle artikli raames tegelema ei hakka. Valisin keeleks groovy. Allpool on testskript, mis lihtsalt inkrementeerib id SOAP sõnumis. On oluline märkida, et te võtate faili flowFile-st, uuendate seda, kuid ärge unustage, et see tuleb uuendatuna tagasi sinna panna. Samuti on oluline märkida, et mitte kõiki raamatukogusid ei ole ühendatud. Võib juhtuda, et peate siiski importima mõne raamatukogu. Puuduseks on ka see, et skripti on sellel kuubikul üsna raske siluda. On võimalus ühendada JVM NIFI-ga ja alustada silumist. Isiklikult käivitasin kohaliku rakenduse ja simuleerisin faili saamist sessioonist. Silumisega tegelesin samuti kohalikult. Vead, mis ilmnevad skripti laadimisel, on üsna lihtsalt guugeldatavad ja NIFI kirjutab need logisse.

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

Kohapeal lõpeb ruudu kohandamine. Seejärel edastatakse uuendatud fail ruudule, mis vastutab faili saatmise eest serverisse. Allpool on esitatud selle ruudu seaded.

Kirjeldame meetodit, mille abil edastatakse SOAP sõnum. Ütleme, kuhu see läheb. Edasi tuleb märkida, et tegemist on just SOAP-iga.

Lisame mõned omadused, näiteks host ja tegevus (soapAction). Salvestame, kontrollime. Täpsemat teavet SOAP päringute saatmise kohta võib vaadata. siin

Oleme arutanud mitmeid NIFI protsesside rakendamisvõimalusi. Kuidas nad omavahel suhestuvad ja milline on nende tegelik kasu. Arutatud näited on testversioonid ja pisut erinevad sellest, mis toimub reaalsetes tingimustes. Loodan, et see artikkel on arendajatele veidi kasulik. Aitäh tähelepanu eest. Kui teil on küsimusi, siis kirjutage. Püüan vastata.

Allikas: habr.com