Apache NIFI — Panoramica delle funzionalità in pratica

Introduzione

È successo che nel mio attuale posto di lavoro ho dovuto familiarizzare con questa tecnologia. Inizierò con una breve introduzione. Durante un incontro, al nostro team è stato detto che era necessario creare un'integrazione con un noto sistema. Per integrazione si intendeva che questo noto sistema ci avrebbe inviato richieste tramite HTTP su un determinato endpoint, e noi, paradossalmente, dovevamo rimandare risposte sotto forma di messaggi SOAP. Sembra tutto semplice e banale. Da questo si evince che è necessario…

Compito

Creare 3 servizi. Il primo di essi è 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, da trasferire al sistema successivo. Viene chiamato l'endpoint del secondo servizio - Servizio di trasporto tramite FTP, che riceve il file trasmesso, lo valida e lo deposita in un storage tramite FTP. Il terzo servizio - Servizio di consegna dati al consumatore, lavora in modo asincrono con i primi due. Accetta una richiesta da un sistema esterno per ricevere il file di cui si parlava sopra, prende il file di risposta pronto, lo modifica (aggiorna i campi id, description, linkToFile) e invia la risposta sotto forma di messaggio SOAP. Quindi, in generale, la situazione è la seguente: i primi due servizi iniziano a lavorare solo quando arrivano dati per l'aggiornamento. Il terzo servizio lavora costantemente poiché ci sono molti consumatori di informazioni, circa 1000 richieste di dati al minuto. I servizi sono costantemente disponibili e le loro istanze si trovano in ambienti diversi, come test, demo, preproduzione e produzione. Di seguito è riportato uno schema di funzionamento di questi servizi. Preciso subito che alcuni dettagli sono semplificati per evitare complicazioni superflue.

Approfondimento tecnico

Nella pianificazione della soluzione del compito, si è deciso prima di realizzare applicazioni in Java utilizzando il framework Spring, con un bilanciatore di carico Nginx, un database Postgres e altre tecnologie, alcune più tecniche e altre meno. Poiché il tempo per elaborare una soluzione tecnica permetteva di considerare approcci alternativi, l'attenzione si è concentrata su una tecnologia alla moda in certi ambienti: Apache NIFI. Devo dire che questa tecnologia ci ha permesso di notare questi 3 servizi. In questo articolo verrà descritta la realizzazione di un servizio di trasporto file e di un servizio di trasmissione dati al consumatore, ma se l'articolo riscuoterà successo, scriverò anche del servizio di aggiornamento dei dati nel database.

Che cos'è

NIFI è un'architettura distribuita per un caricamento e una lavorazione dati rapidi e paralleli, con un gran numero di plugin per fonti e trasformazioni, versioning delle configurazioni e molto altro. Un bonus gradito è che è molto semplice da usare. Processi banali, come getFile, sendHttpRequest e altri, possono essere rappresentati come quadrati. Ogni quadrato rappresenta un certo processo, il cui interazione può essere vista nell'immagine sottostante. Una documentazione più dettagliata sull'interazione e la configurazione dei processi è scritta qui , per chi parla russo — qui. Nella documentazione è spiegato benissimo come estrarre e avviare NIFI, così come come creare processi, anch'essi quadrati.
L'idea di scrivere un articolo è nata dopo lunghe ricerche e strutturazione delle informazioni ricevute in qualcosa di comprensibile, oltre al desiderio di rendere un po' più facile la vita ai futuri sviluppatori..

Esempio

È stato esaminato un esempio di come interagiscono i quadrati tra di loro. Lo schema generale è piuttosto semplice: riceviamo una richiesta HTTP (teoricamente con un file nel corpo della richiesta. Per dimostrare le possibilità di NIFI, in questo esempio la richiesta avvia il processo di acquisizione di un file dal nostro storage locale), quindi inviamo una risposta che indica che la richiesta è stata ricevuta, parallelamente viene avviato il processo di acquisizione del file dallo storage e poi il processo di trasferimento tramite FTP nel nostro storage. Vale la pena chiarire che i processi interagiscono tra di loro tramite il cosiddetto flowFile. Questa è l'entità di base in NIFI, che memorizza attributi e contenuto. Il contenuto è rappresentato dai dati di un file di flusso. In altre parole, se hai ricevuto un file da un quadrato e lo trasferisci a un altro, il contenuto sarà il tuo file.

Come puoi notare, in questa illustrazione è mostrato il processo generale. HandleHttpRequest accetta richieste, ReplaceText genera il corpo della risposta, HandleHttpResponse restituisce la risposta. FetchFile riceve un file dallo storage e lo trasferisce al quadrato PutSftp, che lo carica su FTP all'indirizzo specificato. Ora vediamo più nel dettaglio questo processo.

In questo caso la request rappresenta l'inizio di tutto. Diamo un'occhiata alle sue impostazioni di configurazione.

Qui è tutto piuttosto banale, tranne per StandartHttpContextMap, un servizio che consente di inviare e ricevere richieste. Puoi trovare maggiori dettagli e persino esempi qui — qui

Passiamo ora alle impostazioni di configurazione del quadrato ReplaceText. È importante prestare attenzione a ReplacementValue, che è ciò che verrà restituito all'utente come risposta. Nelle impostazioni puoi regolare il livello di registrazione; puoi visualizzare i log {dove hai estratto nifi}/nifi-1.9.2/logs. Qui ci sono anche parametri di failure/success: basandoti su questi parametri puoi regolare il processo in generale. In caso di elaborazione riuscita del testo, verrà attivato il processo di invio della risposta all'utente, mentre nel caso di esito negativo, registreremo semplicemente il processo non riuscito.

Nelle proprietà di HandleHttpResponse non c'è nulla di particolarmente interessante tranne lo stato in caso di creazione riuscita della risposta.

Dopo aver esaminato richiesta e risposta, passiamo ora a come ottenere un file e caricarlo su un server FTP. FetchFile riceve il file dal percorso specificato nelle impostazioni e lo trasferisce al processo successivo.

Il riquadro PutSftp posiziona il file nell'archivio file. Possiamo vedere di seguito i parametri di configurazione.

È importante notare che ogni riquadro è un processo separato che deve essere avviato. Abbiamo esaminato il caso più semplice che non richiede alcuna personalizzazione complessa. Ora passeremo a un processo leggermente più complicato, dove scriveremo un po' in Groovy.

Esempio più complesso

Il servizio di trasferimento dati al consumatore è risultato un po' più complesso a causa del processo di modifica del messaggio SOAP. Il processo generale è illustrato nell'immagine sottostante.

Qui l'idea non è particolarmente complicata: abbiamo ricevuto una richiesta dal consumatore per i dati, abbiamo inviato una risposta confermando di aver ricevuto il messaggio, abbiamo avviato il processo di ricezione del file di risposta, successivamente lo abbiamo modificato con una logica definita e infine abbiamo inviato il file al consumatore sotto forma di messaggio SOAP al server.

Non credo sia necessario descrivere nuovamente i riquadri già visti sopra: passiamo direttamente ai nuovi. Se hai bisogno di modificare un file e i riquadri standard come ReplaceText non sono adatti, dovrai scrivere il tuo script. Puoi farlo utilizzando il riquadro ExecuteGroovyScript. Le sue impostazioni sono presentate di seguito.

Ci sono due opzioni per caricare lo script in questo quadrato. La prima consiste nel caricare un file contenente lo script. La seconda 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 sentono davvero il bisogno, è necessario creare un quadrato personalizzato e inserirlo nel sistema NIFI. Tutta questa operazione è accompagnata da numerosi passaggi complessi, che non tratteremo in questo articolo. Ho scelto il linguaggio groovy. Di seguito è presentato uno script di prova che aggiorna semplicemente in modo incrementale l'id nel messaggio SOAP. È importante notare che si prende il file da flowFile, lo si aggiorna e non si deve dimenticare di riporlo, aggiornato, allí dove era. Vale la pena notare che non tutte le librerie sono collegate. Potrebbe succedere che sia necessario importare una delle librerie. Un altro svantaggio è che lo script in questo quadrato è piuttosto difficile da fare il debug. Esiste un modo per collegarsi alla JVM NIFI e avviare il processo di debugging. Personalmente, ho avviato un'applicazione locale e ho simulato la ricezione del file dalla sessione. Ho anche svolto il debug localmente. Gli errori che emergono durante il caricamento dello script sono piuttosto facili da cercare su Google e sono registrati direttamente da NIFI 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)
}

In effetti, qui termina la personalizzazione del quadrato. Successivamente, il file aggiornato viene trasferito in un quadrato che si occupa di inviare il file al server. Di seguito è presentata la configurazione di questo quadrato.

Descriviamo il metodo attraverso il quale verrà inviato il messaggio SOAP. Indichiamo dove. Inoltre, è necessario specificare che si tratta di SOAP.

Aggiungiamo alcune proprietà come host e soapAction. Salviamo, verifichiamo. Per ulteriori dettagli su come inviare richieste SOAP, puoi consultare qui

Abbiamo esaminato diverse opzioni per l'utilizzo dei processi NIFI. Come interagiscono e quale sia il loro reale valore. Gli esempi esaminati sono testuali e differiscono leggermente da ciò che accade in produzione. Spero che questo articolo sia utile per gli sviluppatori. Grazie per l'attenzione. Se hai domande, scrivi pure. Cercherò di rispondere.

Fonte: habr.com