Apache NIFI - Una Breve Panoramica di Opportunità in Pratica

Introduzione

Hè accadutu chì in u mo locu di travagliu attuale aghju avutu a cunniscenza di sta tecnulugia. Cumincià cù un pocu di fondu. À a prossima riunione, u nostru squadra hè statu dettu chì avemu bisognu di creà integrazione cù sistema cunnisciutu. Per integrazione si vulia dì chì stu sistema ben cunnisciutu ci mandava dumande via HTTP à un puntu finale specificu, è noi, stranamente, rinviaremu risposte in forma di missaghju SOAP. Tuttu pare simplice è triviale. Da questu ne segue chì avete bisognu ...

Objettivu

Crea 3 servizii. U primu di elli hè u serviziu di l'aghjurnamentu di a basa di dati. Stu serviziu, quandu i novi dati ghjunghjenu da un sistema di terzu, aghjurnà i dati in a basa di dati è genera un schedariu in formatu CSV per trasfirillu à u sistema dopu. L'endpoint di u sicondu serviziu hè chjamatu - u serviziu di trasportu FTP, chì riceve u schedariu trasferitu, u validate, è u mette in l'almacenamiento d'archivi via FTP. U terzu serviziu, u serviziu di trasferimentu di dati di u cunsumadore, travaglia in modu asincronu cù i primi dui. Riceve una dumanda da un sistema esternu di terzu per riceve u schedariu discutitu sopra, piglia u schedariu di risposta pronta, u mudifica (aghjurnà l'id, a descrizzione, i campi linkToFile) è manda a risposta in forma di missaghju SOAP. Questu hè, a stampa generale hè a siguenti: i primi dui servizii cumincianu u so travagliu solu quandu i dati per l'aghjurnamentu sò ghjunti. U terzu serviziu travaglia in permanenza perchè ci sò parechji cunsumatori d'infurmazioni, circa 1000 richieste di dati per minutu. I servizii sò dispunibuli constantemente è i so casi sò situati in ambienti differenti, cum'è test, demo, pre-produzione è prod. Quì sottu hè un diagramma di cumu funziona sti servizii. Lasciami chjarificà subitu chì certi dettagli sò stati simplificati per evità una cumplessità inutile.

Approfondimentu tecnicu

Quandu pianificà una suluzione à u prublema, avemu prima decisu di fà applicazioni in Java cù u framework Spring, Nginx balancer, Postgres database è altre cose tecniche è micca cusì tecniche. Dapoi u tempu di sviluppà una suluzione tecnicu ci hà permessu di cunsiderà altri approcci à scioglie stu prublema, u nostru sguardu cascò nantu à a tecnulugia Apache NIFI, chì hè di moda in certi circles. Dicu subitu chì sta tecnulugia ci hà permessu di nutà sti servizii 3. Questu articulu descriverà u sviluppu di un serviziu di trasportu di fugliale è un serviziu di trasferimentu di dati à u cunsumadore, ma se l'articulu hè utile, scriveraghju nantu à u serviziu per l'aghjurnamentu di dati in a basa di dati.

Qual hè?

NIFI hè una architettura distribuita per carica rapida parallela è trasfurmazioni di dati, un gran numaru di plugins per fonti è trasfurmazioni, versione di cunfigurazioni è assai più. Un bonu bonu hè chì hè assai faciule d'utilizà. I prucessi triviali cum'è getFile, sendHttpRequest è altri ponu esse rapprisintati cum'è quadrati. Ogni quadru rapprisenta un prucessu, l'interazzione di quale pò esse vistu in a figura sottu. A documentazione più dettagliata nantu à l'interazzione di u prucessu di installazione hè stata scritta ccà , per quelli chì parlanu russu - ccà. A ducumentazione descrive perfettamente cumu sbulicà è eseguisce NIFI, è cumu per creà prucessi, cunnisciuti ancu com'è quadrati.
L'idea di scrive un articulu hè natu dopu à una longa ricerca è strutturazione di l'infurmazioni ricevuti in qualcosa di cuscente, è ancu a vuluntà di fà a vita un pocu più faciule per i futuri sviluppatori.

Esempiu:

Un esempiu di cumu i quadrati interagisce cù l'altri hè cunsideratu. U schema generale hè abbastanza sèmplice: Ricevemu una dumanda HTTP (In teoria, cù un schedariu in u corpu di a dumanda. Per dimustrà e capacità di NIFI, in questu esempiu a dumanda principia u prucessu di riceve un schedariu da l'almacenamiento di u schedariu locale. ), dopu avemu rinviatu una risposta chì a dumanda hè stata ricevuta, in parallelu u prucessu di riceve un schedariu da FH è dopu u prucessu di trasfurmà via FTP à FH. Hè vale a pena chjarificà chì i prucessi interagiscenu cù l'altri attraversu u chjamatu flowFile. Questa hè l'entità di basa in NIFI chì guarda l'attributi è u cuntenutu. U cuntenutu hè a dati chì hè rapprisintatu da u schedariu di flussu. Vale à dì, à pocu pressu, se ricevi un schedariu da un quadru è u trasfiriu à un altru, u cuntenutu serà u vostru schedariu.

Comu pudete vede, sta stampa mostra u prucessu generale. HandleHttpRequest - accetta richieste, ReplaceText - genera un corpu di risposta, HandleHttpResponse - manda una risposta. FetchFile - riceve un schedariu da un archiviu di u schedariu, u trasferisce à a piazza PutSftp - mette stu schedariu in FTP, à l'indirizzu specificatu. Avà più nantu à stu prucessu.

In questu casu, a dumanda hè u principiu di tuttu. Fighjemu i so paràmetri di cunfigurazione.

Tuttu quì hè abbastanza triviale cù l'eccezzioni di StandardHttpContextMap - questu hè un tipu di serviziu chì permette di mandà è riceve richieste. In più dettagliu è ancu cù esempii, pudete vede - ccà

Dopu, fighjemu i paràmetri di cunfigurazione ReplaceText di u quadru. Hè vale a pena attente à ReplacementValue - questu hè ciò chì serà tornatu à l'utilizatore in forma di risposta. In i paràmetri pudete aghjustà u livellu di logging, pudete vede i logs {induve avete unpacked nifi}/nifi-1.9.2/logs, ci sò ancu paràmetri di fallimentu / successu - basatu nantu à questi paràmetri pudete regulà u prucessu in tuttu. . Questu hè, in u casu di u prucessu di testu successu, u prucessu di mandà una risposta à l'utilizatore serà chjamatu, è in un altru casu avemu da logu solu u prucessu senza successu.

Ùn ci hè nunda particularmente interessante in e proprietà di HandleHttpResponse eccettu u statutu quandu una risposta hè creata bè.

Avemu risoltu a dumanda è a risposta - andemu à riceve u schedariu è u mette nantu à u servitore FTP. FetchFile - riceve un schedariu à u percorsu specificatu in i paràmetri è u passa à u prossimu prucessu.

E poi u quadru PutSftp - mette u schedariu in u schedariu di almacenamento. Pudemu vede i paràmetri di cunfigurazione sottu.

Hè vale a pena attente à u fattu chì ogni quadru hè un prucessu separatu chì deve esse lanciatu. Avemu vistu l'esempiu più simplice chì ùn hà micca bisognu di persunalizazione cumplessa. In seguitu, fighjemu u prucessu un pocu più cumplicatu, induve scriveremu un pocu nantu à i groove.

Esempiu più cumplessu

U serviziu di trasferimentu di dati à u cunsumadore hè diventatu un pocu più cumplicatu per via di u prucessu di mudificà u missaghju SOAP. U prucessu generale hè mostratu in a figura sottu.

Quì l'idea ùn hè ancu micca particularmente complicata: avemu ricevutu una dumanda da u cunsumadore chì avia bisognu di dati, hà mandatu una risposta chì avia ricevutu un missaghju, hà iniziatu u prucessu di riceve u schedariu di risposta, dopu editatu cù una certa logica, è dopu. trasfirìu u schedariu à u cunsumadore in a forma di un missaghju SOAP à u servitore.

Pensu chì ùn ci hè bisognu di discrive di novu quelli quadrati chì avemu vistu sopra - andemu direttamente à i novi. Sè avete bisognu di edità qualsiasi fugliale è i quadrati di tipu ReplaceText ordinariu ùn sò micca adattati, avete da scrive u vostru propiu script. Questu pò esse fattu cù u quadru ExecuteGroogyScript. I so paràmetri sò presentati quì sottu.

Ci hè duie opzioni per carricà u script in questu quadru. U primu hè di scaricà un schedariu cù un script. U sicondu hè inseritu un script in scriptBody. Quantu sò, u quadru executeScript sustene parechje lingue - una di elle hè groovy. Decederaghju i sviluppatori java - ùn pudete micca scrive script in java in tali quadrati. Per quelli chì volenu veramente, avete bisognu di creà u vostru propiu quadru persunalizatu è aghjunghje à u sistema NIFI. Sta operazione sana hè accumpagnata da un ballu longu cù un tamburinu, chì ùn avemu micca trattatu in questu articulu. Aghju sceltu a lingua groovy. Quì sottu hè un script di prova chì aghjurnà in modu incrementale l'id in un missaghju SOAP. Hè impurtante à nutà. Pigliate u schedariu da flowFile è l'aghjurnà, ùn vi scurdate micca chì avete bisognu di rimette quì, aghjurnatu. Hè nutate ancu chì micca tutte e biblioteche sò incluse. Pò accade chì avete sempre impurtà una di e libbre. Un altru svantaghju hè chì u script in questa piazza hè abbastanza difficiuli di debug. Ci hè un modu per cunnette à u NIFI JVM è inizià u prucessu di debugging. In modu persunale, aghju lanciatu una applicazione locale è simulate riceve un schedariu da a sessione. Aghju ancu fattu debugging in u locu. Errori chì appariscenu quandu caricate un script sò abbastanza faciuli à Google è sò scritti da NIFI stessu à u logu.

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

In fatti, hè quì chì finisce a persunalizazione di u quadru. In seguitu, u schedariu aghjurnatu hè trasferitu à a piazza, chì hè rispunsevule per mandà u schedariu à u servitore. Quì sottu sò i paràmetri di sta piazza.

Descrivemu u metudu per quale un missaghju SOAP serà trasmessu. Scrivemu induve. Dopu avete bisognu di indicà chì questu hè SOAP.

Aghjunghjite parechje proprietà cum'è host è action (soapAction). Salvemu è verificà. Pudete vede più dettagli nantu à cumu mandà dumande SOAP ccà

Avemu vistu parechje opzioni per aduprà prucessi NIFI. Cumu interagiscenu è quale hè u so veru benefiziu? L'esempii cunsiderati sò quelli di prova è sò un pocu sfarente da ciò chì succede veramente in cumbattimentu. Spergu chì questu articulu serà un pocu utile per i sviluppatori. Grazie per a vostra attenzione. Sì avete qualchì quistione, scrivite. Pruvaraghju di risponde.

Source: www.habr.com