Apache NIFI - Kratek pregled funkcij v praksi

Predstavitev

Tako se je zgodilo, da sem se moral na svojem trenutnem delovnem mestu seznaniti s to tehnologijo. Začel bom z malo ozadja. Na naslednjem sestanku so naši ekipi povedali, da moramo ustvariti integracijo s znan sistem. Z integracijo je bilo mišljeno, da nam bo ta dobro znani sistem poslal zahteve prek HTTP na določeno končno točko, mi pa bomo, nenavadno, poslali nazaj odgovore v obliki sporočila SOAP. Vse se zdi preprosto in trivialno. Iz tega sledi, da potrebujete...

Naloga

Ustvarite 3 storitve. Prva med njimi je Database Update Service. Ta storitev, ko novi podatki prispejo iz sistema tretje osebe, posodobi podatke v bazi podatkov in ustvari datoteko v formatu CSV za prenos v naslednji sistem. Pokliče se končna točka druge storitve - FTP Transport Service, ki prenese preneseno datoteko, jo potrdi in shrani v shrambo datotek prek FTP. Tretja storitev, storitev prenosa podatkov potrošnikov, deluje asinhrono s prvima dvema. Od zunanjega sistema tretje osebe prejme zahtevo za sprejem zgoraj obravnavane datoteke, vzame pripravljeno odzivno datoteko, jo spremeni (posodobi ID, opis, polja linkToFile) in pošlje odgovor v obliki sporočila SOAP. To pomeni, da je splošna slika naslednja: prvi dve storitvi začneta delovati šele, ko prispejo podatki za posodobitev. Tretja storitev deluje stalno, ker je porabnikov informacij veliko, približno 1000 zahtevkov za podatke na minuto. Storitve so stalno na voljo in njihovi primerki se nahajajo v različnih okoljih, kot so testno, predstavitveno, predprodukcijsko in proizvodno. Spodaj je diagram delovanja teh storitev. Naj takoj pojasnim, da so nekatere podrobnosti poenostavljene, da bi se izognili nepotrebni zapletenosti.

Tehnično poglabljanje

Ko smo načrtovali rešitev problema, smo se najprej odločili, da bomo aplikacije delali v Javi z uporabo Spring frameworka, Nginx balancerja, Postgres podatkovne baze in drugih tehničnih in manj tehničnih stvari. Ker nam je čas za razvoj tehnične rešitve omogočil razmišljanje o drugih pristopih k reševanju tega problema, nam je pogled padel na v določenih krogih modno tehnologijo Apache NIFI. Takoj bom rekel, da nam je ta tehnologija omogočila, da opazimo te 3 storitve. Ta članek bo opisal razvoj storitve prenosa datotek in storitve prenosa podatkov do potrošnika, če pa bo članek koristen, bom pisal o storitvi za posodabljanje podatkov v bazi.

Kaj pa je to

NIFI je porazdeljena arhitektura za hitro vzporedno nalaganje in obdelavo podatkov, veliko število vtičnikov za vire in transformacije, različice konfiguracij in še veliko več. Lep bonus je, da je zelo enostaven za uporabo. Trivialne procese, kot so getFile, sendHttpRequest in drugi, lahko predstavimo kot kvadratke. Vsak kvadrat predstavlja proces, katerega interakcijo lahko vidimo na spodnji sliki. Napisana je podrobnejša dokumentacija o interakcijah nastavitev procesa tukaj , za tiste, ki govorijo rusko - tukaj. Dokumentacija odlično opisuje, kako razpakirati in zagnati NIFI ter kako ustvariti procese, znane tudi kot kvadrati
Ideja za pisanje članka se je porodila po dolgem iskanju in strukturiranju prejetih informacij v nekaj zavestnega, pa tudi v želji, da bi bodočim razvijalcem nekoliko olajšali življenje.

Primer

Obravnavan je primer medsebojnega delovanja kvadratov. Splošna shema je precej preprosta: prejmemo zahtevo HTTP (teoretično z datoteko v telesu zahteve. Za prikaz zmožnosti NIFI v tem primeru zahteva začne postopek prejema datoteke iz lokalnega pomnilnika datotek ), nato pošljemo nazaj odgovor, da je bila zahteva prejeta, vzporedno poteka postopek prejema datoteke od FH in nato postopek premika le-te preko FTP na FH. Vredno je pojasniti, da procesi medsebojno delujejo prek tako imenovane flowFile. To je osnovna entiteta v NIFI, ki shranjuje atribute in vsebino. Vsebina so podatki, ki jih predstavlja datoteka toka. To je, grobo rečeno, če prejmete datoteko iz enega kvadrata in jo prenesete v drugega, bo vsebina vaša datoteka.

Kot lahko vidite, ta slika prikazuje splošen postopek. HandleHttpRequest - sprejme zahteve, ReplaceText - ustvari telo odgovora, HandleHttpResponse - pošlje odgovor. FetchFile - sprejme datoteko iz pomnilnika datotek, jo prenese na kvadrat PutSftp - postavi to datoteko na FTP, na določen naslov. Zdaj pa več o tem procesu.

V tem primeru je prošnja začetek vsega. Poglejmo njegove konfiguracijske parametre.

Vse tukaj je precej trivialno z izjemo StandardHttpContextMap - to je neke vrste storitev, ki vam omogoča pošiljanje in prejemanje zahtev. Podrobneje in celo s primeri si lahko ogledate - tukaj

Nato si poglejmo konfiguracijske parametre kvadrata ReplaceText. Vredno je biti pozoren na ReplacementValue - to je tisto, kar bo uporabniku vrnjeno v obliki odgovora. V nastavitvah lahko prilagodite stopnjo beleženja, vidite lahko dnevnike {kamor ste razpakirali nifi}/nifi-1.9.2/logs, obstajajo tudi parametri napake/uspeha - na podlagi teh parametrov lahko regulirate proces kot celoto . To pomeni, da bo v primeru uspešne obdelave besedila poklican postopek pošiljanja odgovora uporabniku, v drugem primeru pa bomo neuspešni proces enostavno zabeležili.

V lastnostih HandleHttpResponse ni nič posebej zanimivega, razen stanja, ko je odgovor uspešno ustvarjen.

Razvrstili smo zahtevo in odgovor - preidimo na prejem datoteke in njeno namestitev na strežnik FTP. FetchFile - sprejme datoteko na poti, določeni v nastavitvah, in jo posreduje naslednjemu procesu.

In nato kvadrat PutSftp - postavi datoteko v shrambo datotek. Spodaj lahko vidimo konfiguracijske parametre.

Vredno je biti pozoren na dejstvo, da je vsak kvadrat ločen proces, ki ga je treba zagnati. Ogledali smo si najpreprostejši primer, ki ne zahteva zapletenih prilagoditev. Nato si bomo postopek ogledali malo bolj zapleteno, kjer bomo malo pisali o utorih.

Bolj zapleten primer

Storitev prenosa podatkov do potrošnika se je izkazala za nekoliko bolj zapleteno zaradi postopka spreminjanja sporočila SOAP. Splošni postopek je prikazan na spodnji sliki.

Tudi tukaj ideja ni posebej zapletena: od potrošnika smo prejeli zahtevo, da potrebuje podatke, poslali odgovor, da je prejel sporočilo, sprožili proces prejema odzivne datoteke, jo nato uredili z določeno logiko in nato prenesel datoteko potrošniku v obliki sporočila SOAP na strežnik.

Mislim, da ni potrebe ponovno opisovati tistih kvadratov, ki smo jih videli zgoraj - pojdimo naravnost na nove. Če morate urediti katero koli datoteko in običajni kvadratki tipa ReplaceText niso primerni, boste morali napisati svoj skript. To lahko storite s kvadratom ExecuteGroogyScript. Njegove nastavitve so predstavljene spodaj.

Obstajata dve možnosti za nalaganje skripta v ta kvadrat. Prvi je s prenosom datoteke s skriptom. Drugi je z vstavljanjem skripta v scriptBody. Kolikor vem, executeScript square podpira več jezikov - eden od njih je groovy. Razočaral bom razvijalce java - v takih kvadratih ne morete pisati skriptov v javi. Za tiste, ki to res želite, morate ustvariti svoj kvadrat po meri in ga dodati v sistem NIFI. Vso to operacijo spremlja precej dolg ples s tamburino, ki ga v tem članku ne bomo obravnavali. Izbrala sem groovy jezik. Spodaj je testni skript, ki preprosto postopoma posodablja ID v sporočilu SOAP. Pomembno je opozoriti. Datoteko vzamete iz flowFile in jo posodobite, ne pozabite, da jo morate posodobljeno vrniti tja. Omeniti velja tudi, da niso vključene vse knjižnice. Lahko se zgodi, da boste vseeno morali uvoziti katero od knjižnic. Druga pomanjkljivost je, da je skript v tem kvadratu precej težko odpraviti napake. Obstaja način, da se povežete z NIFI JVM in začnete postopek odpravljanja napak. Osebno sem zagnal lokalno aplikacijo in simuliral prejem datoteke iz seje. Lokalno sem tudi odpravljal napake. Napake, ki se pojavijo pri nalaganju skripta, so Googlu precej enostavne in jih NIFI sam zapiše v dnevnik.

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

Pravzaprav se tu prilagajanje kvadrata konča. Nato se posodobljena datoteka prenese na kvadrat, ki je odgovoren za pošiljanje datoteke na strežnik. Spodaj so nastavitve za ta kvadrat.

Opisujemo metodo, s katero bo posredovano sporočilo SOAP. Pišemo kje. Nato morate označiti, da je to SOAP.

Dodajte več lastnosti, kot sta gostitelj in dejanje (soapAction). Shranjujemo in preverjamo. Ogledate si lahko več podrobnosti o pošiljanju zahtev SOAP tukaj

Ogledali smo si več možnosti za uporabo procesov NIFI. Kako medsebojno delujejo in kakšna je njihova resnična korist? Obravnavani primeri so testni in se nekoliko razlikujejo od tega, kar se dejansko dogaja v boju. Upam, da bo ta članek malo koristen za razvijalce. Hvala za vašo pozornost. Če imate kakšna vprašanja, pišite. Poskusil bom odgovoriti.

Vir: www.habr.com