Sissejuhatus
Juhtus nii, et oma praeguses töökohas pidin selle tehnoloogiaga tutvuma. Alustan väikese taustaga. Järgmisel koosolekul öeldi meie meeskonnale, et peame looma integratsiooni tuntud süsteem. Integratsiooni all peeti silmas seda, et see tuntud süsteem saadaks meile HTTP kaudu päringuid konkreetsesse lõpp-punkti ja meie, kummalisel kombel, saadame vastused tagasi SOAP-teate kujul. Kõik tundub lihtne ja triviaalne. Sellest järeldub, et vajate...
Ülesanne
Looge 3 teenust. Esimene neist on andmebaasi värskendusteenus. See teenus värskendab kolmanda osapoole süsteemist uute andmete saabumisel andmebaasis olevaid andmeid ja genereerib CSV-vormingus faili, mis edastab need järgmisse süsteemi. Teise teenuse lõpp-punkti nimetatakse FTP transporditeenuseks, mis võtab edastatud faili vastu, kinnitab selle ja paneb selle FTP kaudu failimällu. Kolmas teenus, tarbijaandmete edastamise teenus, töötab asünkroonselt kahe esimesega. See võtab vastu kolmanda osapoole välissüsteemi taotluse ülalkirjeldatud faili vastuvõtmiseks, võtab valmis vastusefaili, muudab seda (värskendab ID-d, kirjeldust, linkToFile'i välju) ja saadab vastuse SOAP-teate kujul. See tähendab, et üldpilt on järgmine: kaks esimest teenust alustavad tööd alles siis, kui värskendamiseks vajalikud andmed on saabunud. Kolmas teenus töötab pidevalt, sest infotarbijaid on palju, umbes 1000 andmepäringut minutis. Teenused on pidevalt saadaval ja nende eksemplarid asuvad erinevates keskkondades, nagu test-, demo-, eeltootmis- ja prod. Allpool on diagramm nende teenuste toimimise kohta. Täpsustan kohe, et mõningaid üksikasju on tarbetu keerukuse vältimiseks lihtsustatud.
Tehniline süvendamine
Probleemile lahendust planeerides otsustasime esmalt teha Java-s rakendused, kasutades Spring raamistikku, Nginx balancerit, Postgresi andmebaasi ja muid tehnilisi ja mitte nii tehnilisi asju. Kuna tehnilise lahenduse väljatöötamise aeg võimaldas meil kaaluda selle probleemi lahendamiseks teisi lähenemisviise, langes meie pilk teatud ringkondades moes olevale Apache NIFI tehnoloogiale. Ütlen kohe, et see tehnoloogia võimaldas meil neid 3 teenust märgata. Selles artiklis kirjeldatakse failiedastusteenuse ja andmeedastusteenuse arendamist tarbijale, kuid kui artiklist on kasu, siis kirjutan andmebaasis olevate andmete uuendamise teenusest.
Mis see on
NIFI on hajutatud arhitektuur andmete kiireks paralleelseks laadimiseks ja töötlemiseks, suur hulk pistikprogramme allikate ja teisenduste jaoks, konfiguratsioonide versioonide loomine ja palju muud. Hea boonus on see, et seda on väga lihtne kasutada. Triviaalseid protsesse nagu getFile, sendHttpRequest jt saab esitada ruutudena. Iga ruut tähistab protsessi, mille koosmõju on näha alloleval joonisel. Protsessi seadistamise interaktsioonide kohta on kirjutatud üksikasjalikum dokumentatsioon , neile, kes räägivad vene keelt - . Dokumentatsioon kirjeldab suurepäraselt, kuidas NIFI lahti pakkida ja käivitada ning kuidas luua protsesse, mida nimetatakse ka ruutudeks.
Mõte kirjutada artikkel sündis pärast pikka otsimist ja saadud info millekski teadlikuks struktureerimist, aga ka soovist tulevaste arendajate elu veidi lihtsamaks teha.
Näide
Vaadeldakse näidet, kuidas ruudud üksteisega suhtlevad. Üldskeem on üsna lihtne: Saame HTTP päringu (Teoreetiliselt koos failiga päringu kehas. NIFI võimaluste demonstreerimiseks käivitab selles näites päring faili vastuvõtmise protsessi kohalikust failimälust ), siis saadame tagasi vastuse, et päring on vastu võetud, paralleelselt FH-lt faili vastuvõtmise ja seejärel FTP kaudu FH-sse teisaldamise protsessiga. Tasub selgitada, et protsessid suhtlevad üksteisega nn flowFile'i kaudu. See on NIFI baasüksus, mis salvestab atribuute ja sisu. Sisu on andmed, mida voofail esindab. See tähendab, et jämedalt öeldes, kui saate faili ühest ruudust ja edastate selle teise, on sisu teie fail.
Nagu näete, näitab see pilt üldist protsessi. HandleHttpRequest – võtab päringuid vastu, ReplaceText – genereerib vastuse keha, HandleHttpResponse – saadab vastuse. FetchFile – võtab faili vastu failimälust, kannab selle ruudukujulisse PutSftp – paneb selle faili FTP-le, määratud aadressile. Nüüd sellest protsessist lähemalt.
Sel juhul on taotlus kõige algus. Vaatame selle konfiguratsiooni parameetreid.
Siin on kõik üsna triviaalne, välja arvatud StandardHttpContextMap - see on omamoodi teenus, mis võimaldab teil päringuid saata ja vastu võtta. Üksikasjalikumalt ja isegi näidete abil näete -
Järgmisena vaatame ruudu ReplaceText konfiguratsiooni parameetreid. Tasub pöörata tähelepanu ReplacementValue-le - see tagastatakse kasutajale vastuse kujul. Seadetes saab reguleerida logimise taset, näed logisid {kus sa lahti pakkisid nifi}/nifi-1.9.2/logs, seal on ka ebaõnnestumise/edukuse parameetrid - nende parameetrite alusel saab protsessi tervikuna reguleerida . See tähendab, et eduka tekstitöötluse korral kutsutakse kasutajale vastuse saatmise protsess välja ja teisel juhul lihtsalt logime ebaõnnestunud protsessi.
HandleHttpResponse'i atribuutides pole midagi eriti huvitavat, välja arvatud olek, kui vastus on edukalt loodud.
Oleme päringu ja vastuse lahendanud – jätkame faili vastuvõtmise ja FTP-serverisse paigutamisega. FetchFile - võtab faili vastu seadetes määratud teel ja edastab selle järgmisele protsessile.
Ja siis PutSftp ruut – asetab faili failimällu. Konfiguratsiooniparameetreid näeme allpool.
Tasub pöörata tähelepanu asjaolule, et iga ruut on eraldi protsess, mis tuleb käivitada. Vaatasime kõige lihtsamat näidet, mis ei nõua keerulist kohandamist. Järgmisena vaatame protsessi veidi keerulisemalt, kus kirjutame veidi soonte peale.
Keerulisem näide
Andmeedastusteenus tarbijale osutus SOAP-sõnumi muutmise protsessi tõttu veidi keerulisemaks. Üldine protsess on näidatud alloleval joonisel.
Siin ei ole ka idee eriti keeruline: saime tarbijalt päringu, et ta vajab andmeid, saatsime vastuse, et on saanud sõnumi, alustasime vastusefaili saamise protsessi, seejärel redigeerisime seda teatud loogikaga ja siis edastas faili tarbijale SOAP-sõnumi kujul serverisse.
Ma arvan, et neid ruute, mida eespool nägime, pole vaja uuesti kirjeldada - liigume otse uute juurde. Kui teil on vaja mõnda faili redigeerida ja tavalised ReplaceText tüüpi ruudud ei sobi, peate kirjutama oma skripti. Seda saab teha ExecuteGroogyScripti ruudu abil. Selle seaded on toodud allpool.
Skripti sellele ruudule laadimiseks on kaks võimalust. Esimene on skriptiga faili allalaadimine. Teine on skripti sisestamine scriptBodysse. Minu teada toetab executeScripti ruut mitut keelt - üks neist on groovy. Ma valmistan java arendajatele pettumuse – sellistesse ruutudesse ei saa javas skripte kirjutada. Neile, kes seda tõesti tahavad, peate looma oma kohandatud ruudu ja lisama selle NIFI süsteemi. Kogu seda operatsiooni saadab üsna pikk tants tamburiiniga, mida me selles artiklis ei käsitle. Valisin groovy keele. Allpool on testskript, mis lihtsalt värskendab järk-järgult ID-d SOAP-sõnumis. Oluline on märkida. Võtate faili flowFile'ist ja värskendate seda, ärge unustage, et peate selle värskendatuna sinna tagasi panema. Samuti väärib märkimist, et kõik raamatukogud pole kaasatud. Võib juhtuda, et peate ikkagi mõne libsi importima. Teine miinus on see, et selle ruudu skripti on üsna raske siluda. NIFI JVM-iga ühenduse loomiseks ja silumisprotsessi alustamiseks on võimalus. Isiklikult käivitasin kohaliku rakenduse ja simuleerisin seansist faili saamist. Tegin ka kohapeal silumist. Skripti laadimisel ilmnevad vead on guugeldatavad üsna lihtsalt ja NIFI ise 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)
}Tegelikult sellega ruudu kohandamine lõpeb. Järgmisena kantakse uuendatud fail ruutu, mis vastutab faili serverisse saatmise eest. Allpool on selle ruudu seaded.
Kirjeldame meetodit, mille abil SOAP-sõnum edastatakse. Kirjutame kuhu. Järgmisena peate märkima, et see on SOAP.
Lisage mitu atribuuti, nagu host ja tegevus (soapAction). Salvestame ja kontrollime. Lisateavet SOAP-päringute saatmise kohta näete
Vaatasime NIFI protsesside kasutamiseks mitmeid võimalusi. Kuidas nad omavahel suhtlevad ja mis on nende tegelik kasu? Vaadeldavad näited on katsenäited ja erinevad veidi lahingus tegelikult toimuvast. Loodan, et see artikkel on arendajatele pisut kasulik. Tänan tähelepanu eest. Küsimuste korral kirjutage. Püüan vastata.
Allikas: www.habr.com