Introduzione
A lavoro, mi è capitato di dover utilizzare questa tecnologia. Iniziamo con una breve introduzione. Durante una riunione, ci è stato comunicato che dovevamo creare un'integrazione con un sistema noto. Per integrazione si intendeva che questo sistema noto ci avrebbe inviato richieste tramite HTTP a un determinato endpoint, e noi, in modo curioso, avremmo inviato indietro risposte sotto forma di messaggio SOAP. Sembra tutto semplice e banale. Da ciò segue che dobbiamo...
Compito
Creare 3 servizi. Il primo è il Servizio di aggiornamento del database. Questo servizio, all'arrivo di nuovi dati da un sistema esterno, aggiorna i dati nel database e genera un file in formato CSV, per la sua trasmissione al sistema successivo. Viene chiamato l'endpoint del secondo servizio - il Servizio di trasporto tramite FTP, che riceve il file trasmesso, lo convalida e lo archivia nello storage tramite FTP. Il terzo servizio - il Servizio di trasmissione dei dati al consumatore, lavora in modo asincrono con i primi due. Riceve una richiesta da un sistema esterno per ottenere il file di cui sopra, prende il file di risposta pronto, lo modifica (aggiornando i campi id, description, linkToFile) e invia la risposta come messaggio SOAP. Quindi, in generale, la situazione è la seguente: i primi due servizi iniziano il loro lavoro solo quando arrivano dati per l'aggiornamento. Il terzo servizio lavora continuamente poiché ci sono molti consumatori di informazioni, circa 1000 richieste di dati al minuto. I servizi sono sempre disponibili e le loro istanze si trovano in ambienti diversi, come test, demo, pre-produzione e produzione. Di seguito è riportato un diagramma di funzionamento di questi servizi. Aggiungo subito che alcuni dettagli sono stati semplificati per evitare complessità eccessive.
Approfondimento tecnico
Nella pianificazione della soluzione, inizialmente abbiamo deciso di creare un'applicazione in Java utilizzando il framework Spring, con un bilanciatore Nginx, un database Postgres e altre tecnologie. Poiché il tempo a disposizione per sviluppare la soluzione tecnica ci permetteva di considerare altre possibile approcci, abbiamo rivolto la nostra attenzione alla tecnologia alla moda in determinati ambienti, Apache NIFI. Devo dire che questa tecnologia ci ha permesso di notare questi 3 servizi. In questo articolo descriverò lo sviluppo del servizio di trasporto file e del servizio di trasmissione dati al consumatore; però, se l'articolo sarà ben accolto, scriverò anche del servizio di aggiornamento dei dati nel database.
Cos'è?
NIFI è un'architettura distribuita per il caricamento e l'elaborazione rapida dei dati, con un'ampia gamma di plugin per fonti e trasformazioni, versionamento delle configurazioni e molto altro. Un bel vantaggio è che è molto semplice da usare. I processi banali, come getFile, sendHttpRequest e altri, possono essere rappresentati come riquadri. Ogni riquadro rappresenta un processo, la cui interazione può essere vista nell'immagine sottostante. Maggiori dettagli sulla documentazione riguardante l'interazione e la configurazione dei processi sono scritti , per chi parla russo — . Nella documentazione è spiegato benissimo come estrarre e avviare NIFI, e come creare processi, che sono appunto i riquadri
L'idea di scrivere un articolo è nata dopo prolungate ricerche e la strutturazione delle informazioni ottenute in qualcosa di coerente, oltre al desiderio di rendere un po' più facile la vita ai futuri sviluppatori.
Esempio
Viene esaminato un esempio di come interagiscono i quadrati tra loro. Lo schema generale è piuttosto semplice: riceviamo una richiesta HTTP (teoricamente con un file nel corpo della richiesta. Per dimostrare le capacità di NIFI, in questo esempio la richiesta avvia il processo di acquisizione del file dal FХ locale), successivamente inviamo indietro una risposta che conferma che la richiesta è stata ricevuta, mentre viene avviato in parallelo il processo di acquisizione del file dal FХ e il processo di trasferimento tramite FTP nel FХ. È importante sottolineare che i processi interagiscono tra loro tramite la cosiddetta flowFile. Questa è l'entità di base in NIFI che contiene attributi e contenuti. Il contenuto è rappresentato dai dati del file del flusso. In altre parole, se ricevi un file da un quadrato e lo trasferisci a un altro, il contenuto sarà il tuo file.
Come puoi notare, in questa immagine è rappresentato il processo generale. HandleHttpRequest accetta le richieste, ReplaceText genera il corpo della risposta, HandleHttpResponse restituisce la risposta. FetchFile ottiene il file dallo storage e lo passa al quadrato PutSftp, che inserisce il file sull'FTP all'indirizzo specificato. Ora approfondiamo questo processo.
In questo caso, la richiesta è l'inizio di tutto. Diamo un'occhiata alle sue impostazioni di configurazione.
Qui tutto è piuttosto banale, tranne per StandartHttpContextMap — un servizio che permette di inviare e ricevere richieste. Maggiori dettagli, anche con esempi, possono essere consultati qui —
Successivamente, esaminiamo le impostazioni di configurazione di ReplaceText. È importante prestare attenzione a ReplacementValue — ciò che sarà restituito all'utente come risposta. In settings, è possibile regolare il livello di logging, i log possono essere visionati in {куда распаковали nifi}/nifi-1.9.2/logs, dove ci sono anche parametri di failure/success — sulla base di questi parametri, si può regolare l'intero processo. Cioè, in caso di elaborazione del testo riuscita, verrà avviato il processo di invio della risposta all'utente, mentre in caso contrario semplicemente registreremo il processo non riuscito.
Nelle proprietà di HandleHttpResponse non c'è nulla di particolarmente interessante, tranne lo stato in caso di creazione riuscita della risposta.
Abbiamo chiarito la questione della richiesta e della risposta — procediamo ora a ottenere un file e posizionarlo sul server FTP. FetchFile — acquisisce un file dal percorso specificato nelle impostazioni e lo passa al processo successivo.
E poi il quadrato PutSftp inserisce il file nello storage. Possiamo vedere le opzioni di configurazione di seguito.
È importante notare che ogni quadrato è un processo separato che deve essere avviato. Abbiamo esaminato il caso più semplice che non richiede alcuna personalizzazione complessa. Adesso considereremo un processo leggermente più complicato, in cui scriveremo un po' su Groovy.
Esempio più complesso
Il servizio di trasferimento dati al consumatore è diventato un po' più complicato a causa del processo di modifica del messaggio SOAP. Il processo generale è rappresentato nell'immagine qui sotto.
L'idea qui non è particolarmente complessa: abbiamo ricevuto una richiesta dal consumatore riguardo ai dati, abbiamo inviato una risposta che conferma la ricezione del messaggio, abbiamo avviato il processo di ricezione del file di risposta, quindi lo abbiamo modificato con una logica specifica, dopo di che abbiamo inviato il file al consumatore sotto forma di messaggio SOAP al server.
Non credo sia necessario descrivere nuovamente i quadrati che abbiamo visto sopra — passiamo direttamente ai nuovi. Se devi modificare un file e i normali quadrati come ReplaceText non sono adatti, dovrai scrivere il tuo script. Puoi farlo utilizzando il quadrato ExecuteGroogyScript. Le sue impostazioni sono riportate di seguito.
Ci sono due modi per caricare uno script in questo quadrato. Il primo è caricare un file con lo script. Il secondo consiste nell'incollare lo script in scriptBody. Per quanto ne so, il quadrato executeScript supporta diversi linguaggi di programmazione, uno dei quali è Groovy. Deluderò gli sviluppatori Java: non è possibile scrivere script in questi quadrati utilizzando Java. Per coloro che ne hanno davvero voglia, è necessario creare un proprio quadrato personalizzato e aggiungerlo al sistema NIFI. Tutta questa operazione richiede diversi passaggi complessi, di cui non ci occuperemo in questo articolo. Ho scelto il linguaggio Groovy. Di seguito è fornito uno script di test che aggiorna in modo incrementale l'id nel messaggio SOAP. È importante notare che si prende un file da flowFile, lo si aggiorna e non si deve dimenticare di rimetterlo lì, una volta aggiornato. Vale anche la pena sottolineare che non tutte le librerie sono collegate. Potrebbe succedere che dovrete comunque importare una delle librerie. Un altro svantaggio è che lo script in questo quadrato è piuttosto difficile da fare debugging. Esiste un modo per connettersi alla JVM NIFI e avviare il processo di debug. Personalmente, ho eseguito un'applicazione locale e ho simulato il recupero di un file dalla sessione. Ho anche fatto debugging localmente. Gli errori che si verificano durante il caricamento dello script si possono facilmente cercare online e NIFI li registra nel log.
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)
}La personalizzazione di questo quadrato finisce qui. Il file aggiornato viene poi inviato al quadrato che si occupa dell'invio del file al server. Di seguito sono riportate le impostazioni di questo quadrato.
Descriviamo il metodo che verrà utilizzato per inviare il messaggio SOAP. Indichiamo dove. È necessario specificare che si tratta di un messaggio SOAP.
Aggiungiamo diverse proprietà come l'host e l'azione (soapAction). Salviamo e controlliamo. Per ulteriori dettagli su come inviare richieste SOAP, è possibile consultare.
Abbiamo esaminato diverse applicazioni dei processi NIFI. Come interagiscono e quali sono i reali vantaggi che ne derivano. Gli esempi esaminati sono testuali e differiscono leggermente da quelli utilizzati in scenari reali. Spero che questo articolo possa essere utile per gli sviluppatori. Grazie per l'attenzione. Se avete domande, non esitate a scrivere. Cercherò di rispondere.
Fonte: habr.com
