Apache NIFI - Kratak pregled karakteristika u praksi

Uvod

Desilo se da sam na svom sadašnjem radnom mestu morao da se upoznam sa ovom tehnologijom. Počeću sa malo pozadine. Na sljedećem sastanku, našem timu je rečeno da moramo napraviti integraciju sa poznati sistem. Pod integracijom se podrazumijevalo da će nam ovaj dobro poznati sistem slati zahtjeve putem HTTP-a do određene krajnje tačke, a mi bismo, začudo, vraćali odgovore u obliku SOAP poruke. Sve izgleda jednostavno i trivijalno. Iz ovoga proizilazi da vam je potrebno...

Cilj

Kreirajte 3 usluge. Prvi od njih je usluga ažuriranja baze podataka. Ova usluga, kada stignu novi podaci sa sistema treće strane, ažurira podatke u bazi podataka i generiše datoteku u CSV formatu za prenos na sledeći sistem. Krajnja tačka drugog servisa se zove - FTP transportna usluga, koja prima prenesenu datoteku, provjerava je i stavlja je u skladište datoteka putem FTP-a. Treća usluga, usluga prijenosa podataka potrošača, radi asinhrono s prve dvije. Prima zahtjev od vanjskog sistema treće strane da primi datoteku o kojoj je bilo riječi, uzima datoteku spremnog odgovora, modificira je (ažurira id, opis, linkToFile polja) i šalje odgovor u obliku SOAP poruke. Odnosno, ukupna slika je sljedeća: prva dva servisa počinju sa radom tek kada stignu podaci za ažuriranje. Treći servis radi konstantno jer ima mnogo potrošača informacija, oko 1000 zahtjeva za podacima u minuti. Usluge su stalno dostupne i njihove instance se nalaze u različitim okruženjima, kao što su test, demo, predprodukcija i produkcija. Ispod je dijagram kako ove usluge rade. Odmah da pojasnim da su neki detalji pojednostavljeni kako bi se izbjegla nepotrebna složenost.

Technical Deepening

Kada smo planirali rješenje problema, prvo smo se odlučili za izradu aplikacija u Javi koristeći Spring framework, Nginx balanser, Postgres bazu podataka i druge tehničke i ne baš tehničke stvari. Budući da nam je vrijeme za razvoj tehničkog rješenja omogućilo da razmotrimo i druge pristupe rješavanju ovog problema, naš je pogled pao na Apache NIFI tehnologiju, koja je u određenim krugovima moderna. Odmah ću reći da nam je ova tehnologija omogućila da primijetimo ove 3 usluge. Ovaj članak će opisati razvoj usluge prijenosa datoteka i usluge prijenosa podataka do potrošača, ali ako je članak koristan, pisat ću o servisu za ažuriranje podataka u bazi podataka.

Šta je to?

NIFI je distribuirana arhitektura za brzo paralelno učitavanje i obradu podataka, veliki broj dodataka za izvore i transformacije, verzioniranje konfiguracija i još mnogo toga. Dobar bonus je što je vrlo jednostavan za korištenje. Trivijalni procesi kao što su getFile, sendHttpRequest i drugi mogu biti predstavljeni kao kvadrati. Svaki kvadrat predstavlja proces čija se interakcija može vidjeti na slici ispod. Napisana je detaljnija dokumentacija o interakcijama postavljanja procesa ovdje , za one koji govore ruski - ovdje. Dokumentacija savršeno opisuje kako raspakirati i pokrenuti NIFI, kao i kako kreirati procese, također poznate kao kvadrati
Ideja da se napiše članak rodila se nakon dugog traženja i strukturiranja dobijenih informacija u nešto svjesno, kao i želje da se budućim programerima malo olakša život.

Primjer:

Razmatran je primjer kako kvadrati međusobno djeluju. Opća šema je prilično jednostavna: primamo HTTP zahtjev (u teoriji, sa datotekom u tijelu zahtjeva. Da bi demonstrirali mogućnosti NIFI-a, u ovom primjeru zahtjev pokreće proces primanja datoteke iz lokalne memorije datoteka ), zatim šaljemo natrag odgovor da je zahtjev primljen, paralelno s procesom prijema fajla od FH-a i zatim procesom njegovog premještanja preko FTP-a u FH. Vrijedi pojasniti da procesi međusobno komuniciraju kroz takozvani flowFile. Ovo je osnovni entitet u NIFI koji pohranjuje atribute i sadržaj. Sadržaj su podaci koji su predstavljeni stream fajlom. To jest, grubo govoreći, ako dobijete datoteku iz jednog kvadrata i prenesete je u drugi, sadržaj će biti vaš fajl.

Kao što vidite, ova slika prikazuje opći proces. HandleHttpRequest - prihvata zahtjeve, ReplaceText - generiše tijelo odgovora, HandleHttpResponse - šalje odgovor. FetchFile - prima datoteku iz skladišta datoteka, prenosi je u kvadrat PutSftp - stavlja ovu datoteku na FTP, na navedenu adresu. Sada više o ovom procesu.

U ovom slučaju, zahtjev je početak svega. Pogledajmo njegove konfiguracijske parametre.

Ovdje je sve prilično trivijalno s izuzetkom StandardHttpContextMap - ovo je vrsta usluge koja vam omogućava slanje i primanje zahtjeva. Detaljnije, pa čak i sa primjerima, možete vidjeti - ovdje

Zatim, pogledajmo ReplaceText konfiguracijske parametre kvadrata. Vrijedno je obratiti pažnju na ReplacementValue - to je ono što će biti vraćeno korisniku u obliku odgovora. U postavkama možete podesiti nivo evidentiranja, možete vidjeti logove {gdje ste raspakirali nifi}/nifi-1.9.2/logove, tu su i parametri neuspjeha/uspjeha - na osnovu ovih parametara možete regulisati proces u cjelini . Odnosno, u slučaju uspješne obrade teksta, proces slanja odgovora korisniku će biti pozvan, au drugom slučaju ćemo jednostavno evidentirati neuspješan proces.

Ne postoji ništa posebno zanimljivo u svojstvima HandleHttpResponse osim statusa kada je odgovor uspješno kreiran.

Sredili smo zahtjev i odgovor - idemo na prijem fajla i njegovo postavljanje na FTP server. FetchFile - prima datoteku na stazi navedenoj u postavkama i prosljeđuje je sljedećem procesu.

A zatim PutSftp kvadrat - postavlja datoteku u skladište datoteka. U nastavku možemo vidjeti konfiguracijske parametre.

Vrijedno je obratiti pažnju na činjenicu da je svaki kvadrat zaseban proces koji se mora pokrenuti. Pogledali smo najjednostavniji primjer koji ne zahtijeva nikakvu složenu prilagodbu. Zatim ćemo proces pogledati malo kompliciranije, gdje ćemo malo pisati o žljebovima.

Složeniji primjer

Usluga prijenosa podataka do potrošača se pokazala malo složenijom zbog procesa modifikacije SOAP poruke. Opšti proces je prikazan na donjoj slici.

Ovdje ideja također nije posebno komplicirana: primili smo zahtjev od potrošača da su mu potrebni podaci, poslali odgovor da je primio poruku, pokrenuli proces prijema datoteke odgovora, zatim je uredili određenom logikom, a zatim prenio datoteku potrošaču u obliku SOAP poruke na server.

Mislim da nema potrebe ponovno opisivati ​​one kvadrate koje smo vidjeli gore - prijeđimo odmah na nove. Ako trebate urediti bilo koju datoteku, a obični kvadrati tipa ReplaceText nisu prikladni, morat ćete napisati vlastitu skriptu. Ovo se može učiniti pomoću kvadrata ExecuteGroogyScript. Njegove postavke su predstavljene u nastavku.

Postoje dvije opcije za učitavanje skripte u ovaj kvadrat. Prvi je preuzimanjem datoteke sa skriptom. Drugi je umetanjem skripte u scriptBody. Koliko ja znam, kvadrat executeScript podržava nekoliko jezika - jedan od njih je groovy. Razočarat ću Java programere - ne možete pisati skripte u Javi u takvim kvadratima. Za one koji to zaista žele, potrebno je da kreirate svoj sopstveni kvadrat i dodate ga u NIFI sistem. Cijelu ovu operaciju prati prilično dug ples s tamburom, o čemu se nećemo baviti u ovom članku. Izabrao sam groovy jezik. Ispod je testna skripta koja jednostavno postepeno ažurira id u SOAP poruci. Važno je napomenuti. Uzimate datoteku iz flowFile-a i ažurirate je, ne zaboravite da je trebate vratiti tamo, ažuriranu. Također je vrijedno napomenuti da nisu uključene sve biblioteke. Može se desiti da ipak morate da uvezete jednu od biblioteka. Još jedan nedostatak je što je skriptu u ovom kvadratu prilično teško otkloniti. Postoji način da se povežete na NIFI JVM i pokrenete proces otklanjanja grešaka. Lično sam pokrenuo lokalnu aplikaciju i simulirao primanje fajla sa sesije. Također sam radio i lokalno ispravljanje grešaka. Greške koje se pojavljuju prilikom učitavanja skripte su prilično jednostavne za Google i NIFI ih sam upisuje u 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)
}

Zapravo, tu se završava prilagođavanje kvadrata. Zatim se ažurirana datoteka prenosi na kvadrat, koji je odgovoran za slanje datoteke na server. Ispod su postavke za ovaj kvadrat.

Opisujemo metod kojim će se SOAP poruka prenijeti. Pišemo gde. Zatim morate naznačiti da je ovo SOAP.

Dodajte nekoliko svojstava kao što su host i akcija (soapAction). Čuvamo i provjeravamo. Možete vidjeti više detalja o slanju SOAP zahtjeva ovdje

Razmotrili smo nekoliko opcija za korištenje NIFI procesa. Kako su u interakciji i koja je njihova stvarna korist? Razmatrani primjeri su probni i malo se razlikuju od onoga što se stvarno dešava u borbi. Nadam se da će ovaj članak biti malo koristan za programere. Hvala vam na pažnji. Ako imate pitanja, pišite. Pokušaću da odgovorim.

izvor: www.habr.com