įvedimas
Taip jau susiklostė, kad dabartinėje darbo vietoje teko susipažinti su šia technologija. Pradėsiu nuo nedidelio pagrindo. Kitame susitikime mūsų komandai buvo pasakyta, kad turime sukurti integraciją su žinoma sistema. Integracija turėjo omenyje tai, kad ši gerai žinoma sistema siųs mums užklausas per HTTP į konkretų galinį tašką, o mes, kaip bebūtų keista, atsakymus siųstume SOAP žinutės forma. Viskas atrodo paprasta ir nereikšminga. Iš to išplaukia, kad jums reikia...
Užduotis
Sukurkite 3 paslaugas. Pirmasis iš jų yra duomenų bazės atnaujinimo paslauga. Ši paslauga, kai gaunami nauji duomenys iš trečiosios šalies sistemos, atnaujina duomenis duomenų bazėje ir sugeneruoja failą CSV formatu, kad jį perkeltų į kitą sistemą. Antrosios paslaugos galutinis taškas vadinamas - FTP transportavimo tarnyba, kuri priima perduotą failą, jį patvirtina ir per FTP įkelia į failų saugyklą. Trečioji paslauga – vartotojų duomenų perdavimo paslauga – veikia asinchroniškai su pirmosiomis dviem. Jis gauna užklausą iš trečiosios šalies išorinės sistemos gauti aukščiau aptartą failą, paima paruoštą atsakymo failą, jį modifikuoja (atnaujina ID, aprašymą, linkToFile laukus) ir išsiunčia atsakymą SOAP pranešimo forma. Tai yra, bendras vaizdas yra toks: pirmosios dvi paslaugos pradeda veikti tik tada, kai gaunami naujinimui skirti duomenys. Trečioji paslauga veikia nuolat, nes yra daug informacijos vartotojų, apie 1000 duomenų užklausų per minutę. Paslaugos pasiekiamos nuolat, o jų egzemplioriai yra skirtingose aplinkose, pavyzdžiui, bandomojoje, demonstracinėje, išankstinio gamybos ir produkcijos. Žemiau pateikiama diagrama, kaip šios paslaugos veikia. Iš karto paaiškinsiu, kad kai kurios detalės buvo supaprastintos, kad būtų išvengta nereikalingo sudėtingumo.
Techninis gilinimas
Planuodami problemos sprendimą, pirmiausia nusprendėme aplikacijas daryti Java naudojant Spring framework, Nginx balansyrą, Postgres duomenų bazę ir kitus techninius ir ne tokius techninius dalykus. Kadangi techninio sprendimo kūrimo laikas leido apsvarstyti kitus šios problemos sprendimo būdus, mūsų žvilgsnis nukrypo į tam tikruose sluoksniuose madingą technologiją Apache NIFI. Iš karto pasakysiu, kad ši technologija leido mums pastebėti šias 3 paslaugas. Šiame straipsnyje bus aprašyta failų transportavimo paslaugos ir duomenų perdavimo vartotojui paslaugos kūrimas, tačiau jei straipsnis bus naudingas, parašysiu apie duomenų atnaujinimo paslaugą duomenų bazėje.
Kas tai
NIFI yra paskirstyta architektūra, skirta greitai lygiagrečiai įkelti ir apdoroti duomenis, daug šaltinių ir transformacijų įskiepių, konfigūracijų versijų kūrimo ir daug daugiau. Puiki premija yra tai, kad juo labai paprasta naudotis. Trivialūs procesai, tokie kaip getFile, sendHttpRequest ir kiti, gali būti pavaizduoti kaip kvadratai. Kiekvienas kvadratas reiškia procesą, kurio sąveiką galima pamatyti paveikslėlyje žemiau. Buvo parašyta išsamesnė dokumentacija apie proceso sąrankos sąveiką , kalbantiems rusiškai - . Dokumentacijoje puikiai aprašyta, kaip išpakuoti ir paleisti NIFI, taip pat kaip sukurti procesus, dar vadinamus kvadratais.
Idėja parašyti straipsnį gimė po ilgų paieškų ir gautą informaciją susisteminus į kažką sąmoningo, taip pat noro šiek tiek palengvinti gyvenimą būsimiems kūrėjams.
Pavyzdys
Apsvarstytas pavyzdys, kaip kvadratai sąveikauja vienas su kitu. Bendra schema gana paprasta: Gauname HTTP užklausą (Teoriškai su failu užklausos turinyje. Norėdami parodyti NIFI galimybes, šiame pavyzdyje užklausa pradeda failo gavimo iš vietinės failų saugyklos procesą ), tada siunčiame atgal atsakymą, kad užklausa gauta, lygiagrečiai vykstant failo gavimo iš FH ir tada perkeliant jį per FTP į FH procesą. Verta paaiškinti, kad procesai sąveikauja tarpusavyje per vadinamąjį srauto failą. Tai yra pagrindinis NIFI objektas, kuriame saugomi atributai ir turinys. Turinys – tai srauto faile pateikiami duomenys. Tai yra, grubiai tariant, jei gausite failą iš vieno kvadrato ir perkelsite jį į kitą, turinys bus jūsų failas.
Kaip matote, šiame paveikslėlyje parodytas bendras procesas. HandleHttpRequest – priima užklausas, ReplaceText – generuoja atsakymo turinį, HandleHttpResponse – siunčia atsakymą. FetchFile - gauna failą iš failų saugyklos, perkelia jį į kvadratinį PutSftp - įdeda šį failą į FTP, nurodytu adresu. Dabar daugiau apie šį procesą.
Šiuo atveju prašymas yra visko pradžia. Pažvelkime į jo konfigūracijos parametrus.
Viskas čia yra gana nereikšminga, išskyrus „StandardHttpContextMap“ - tai tam tikra paslauga, leidžianti siųsti ir gauti užklausas. Išsamiau ir net su pavyzdžiais galite pamatyti -
Toliau pažiūrėkime į kvadrato ReplaceText konfigūracijos parametrus. Verta atkreipti dėmesį į ReplacementValue – būtent tai bus grąžinta vartotojui atsakymo forma. Nustatymuose galite reguliuoti registravimo lygį, galite matyti žurnalus {kur išpakavote nifi}/nifi-1.9.2/logs, taip pat yra nesėkmės / sėkmės parametrai - pagal šiuos parametrus galite reguliuoti procesą kaip visumą . Tai yra, sėkmingo teksto apdorojimo atveju bus iškviestas atsakymo siuntimo vartotojui procesas, o kitu atveju mes tiesiog užregistruosime nesėkmingą procesą.
HandleHttpResponse ypatybėse nėra nieko ypač įdomaus, išskyrus būseną, kai atsakymas sėkmingai sukurtas.
Sutvarkėme užklausą ir atsakymą – pereikime prie failo gavimo ir įdėjimo į FTP serverį. FetchFile – priima failą nustatymuose nurodytu keliu ir perduoda jį kitam procesui.
Ir tada PutSftp kvadratas - įdeda failą į failų saugyklą. Žemiau matome konfigūracijos parametrus.
Verta atkreipti dėmesį į tai, kad kiekvienas kvadratas yra atskiras procesas, kurį reikia pradėti. Mes pažvelgėme į paprasčiausią pavyzdį, kuriam nereikia sudėtingo pritaikymo. Toliau mes pažvelgsime į procesą šiek tiek sudėtingiau, kur šiek tiek parašysime griovelius.
Sudėtingesnis pavyzdys
Duomenų perdavimo paslauga vartotojui pasirodė kiek sudėtingesnė dėl SOAP pranešimo modifikavimo proceso. Bendras procesas parodytas paveikslėlyje žemiau.
Čia idėja taip pat nėra itin sudėtinga: gavome vartotojo užklausą, kad jam reikia duomenų, išsiuntėme atsakymą, kad gavo žinutę, pradėjome atsakymo failo gavimo procesą, tada redagavome jį su tam tikra logika ir tada. perdavė failą vartotojui SOAP pranešimo į serverį forma.
Manau, nereikia dar kartą apibūdinti tų kvadratų, kuriuos matėme aukščiau - pereikime tiesiai prie naujų. Jei jums reikia redaguoti bet kurį failą ir įprasti ReplaceText tipo kvadratai netinka, turėsite parašyti savo scenarijų. Tai galima padaryti naudojant ExecuteGroogyScript kvadratą. Jo nustatymai pateikti žemiau.
Yra du variantai, kaip įkelti scenarijų į šią aikštę. Pirmasis yra atsisiųsti failą su scenarijumi. Antrasis yra įterpiant scenarijų į scriptBody. Kiek aš žinau, „executeScript“ kvadratas palaiko kelias kalbas - viena iš jų yra groovy. Nuvilsiu java kūrėjus - tokiuose kvadratuose negalima rašyti scenarijų java. Tiems, kurie tikrai nori, turite susikurti savo pasirinktinį kvadratą ir įtraukti jį į NIFI sistemą. Visą šią operaciją lydi gana ilgas šokis su tamburinu, kurio šiame straipsnyje nenagrinėsime. Pasirinkau groovy kalbą. Žemiau yra bandomasis scenarijus, kuris tiesiog laipsniškai atnaujina ID SOAP pranešime. Svarbu pažymėti. Paimkite failą iš „flowFile“ ir atnaujinkite, nepamirškite, kad turite jį įdėti atgal, atnaujintą. Taip pat verta paminėti, kad įtrauktos ne visos bibliotekos. Gali atsitikti taip, kad vis tiek turėsite importuoti vieną iš libs. Kitas minusas yra tai, kad scenarijų šioje aikštėje gana sunku derinti. Yra būdas prisijungti prie NIFI JVM ir pradėti derinimo procesą. Asmeniškai aš paleidau vietinę programą ir imitavau failo gavimą iš sesijos. Taip pat derinau vietoje. Klaidos, atsirandančios įkeliant scenarijų, yra gana nesunkios „Google“ ir jas įrašo pats NIFI į žurnalą.
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)
}Tiesą sakant, čia baigiasi aikštės pritaikymas. Tada atnaujintas failas perkeliamas į kvadratą, kuris yra atsakingas už failo siuntimą į serverį. Žemiau pateikiami šio kvadrato nustatymai.
Aprašome būdą, kuriuo bus perduodamas SOAP pranešimas. Rašome kur. Toliau turite nurodyti, kad tai SOAP.
Pridėkite kelias ypatybes, pvz., pagrindinį kompiuterį ir veiksmą (soapAction). Išsaugome ir tikriname. Galite pamatyti daugiau informacijos apie tai, kaip siųsti SOAP užklausas
Išnagrinėjome keletą NIFI procesų naudojimo galimybių. Kaip jie sąveikauja ir kokia jų tikroji nauda? Nagrinėjami pavyzdžiai yra bandomieji ir šiek tiek skiriasi nuo to, kas iš tikrųjų vyksta kovoje. Tikiuosi, kad šis straipsnis bus šiek tiek naudingas kūrėjams. Ačiū už dėmesį. Jei turite klausimų rašykite. pabandysiu atsakyti.
Šaltinis: www.habr.com