Ciao a tutti!

Il compito consiste nel seguente — c'è un flow, presentato nell'immagine sopra, che deve essere distribuito su N server con Apache NiFi. Il flow è testuale — genera un file e lo invia a un'altra istanza di NiFi. Il trasferimento dei dati avviene tramite il protocollo NiFi Site to Site.

NiFi Site to Site (S2S) è un modo sicuro e facile da configurare per trasferire dati tra le istanze di NiFi. Come funziona S2S, vedere documentazione ed è importante non dimenticare di configurare l'istanza di NiFi per permettere S2S, vedere qui.

Nei casi in cui si parla di trasferimento di dati tramite S2S, un'istanza è chiamata cliente e l'altra server. Il cliente invia i dati, il server li riceve. Ci sono due modi per configurare il trasferimento dei dati tra di loro:

1. Push. Dall'istanza cliente, i dati vengono inviati tramite Remote Process Group (RPG). Nell'istanza server, i dati vengono ricevuti tramite Input Port
2. Pull. Il server riceve i dati tramite RPG, il cliente invia tramite Output port.

Il flow per la distribuzione è memorizzato in Apache Registry.

Apache NiFi Registry è un sotto-progetto di Apache NiFi che fornisce uno strumento per memorizzare flow e gestire versioni. Una sorta di GIT. Le informazioni sull'installazione, configurazione e utilizzo del registry possono essere trovate in documentazione ufficiale. Il flow da memorizzare è raggruppato in un process group e in questo modo è conservato nel registry. Torneremo su questo più avanti nell'articolo.

All'inizio, quando N è un numero ridotto, il flow viene consegnato e aggiornato manualmente in un tempo accettabile.

Ma con l'aumento di N, ci sono più problemi:

1. l'aggiornamento del flow richiede più tempo. È necessario accedere a tutti i server
2. ci sono errori nell'aggiornamento dei template. Qui è stato aggiornato, ma qui è stato dimenticato
3. errori umani durante l'esecuzione di un gran numero di operazioni simili

Tutto ciò ci porta alla conclusione che è necessario automatizzare il processo. Ho provato i seguenti modi per risolvere questo problema:

1. Utilizzare MiNiFi invece di NiFi
2. NiFi CLI
3. NiPyAPI

Utilizzo di MiNiFi

Apache MiNiFy — sottoprogetto Apache NiFi. MiNiFy — un agente compatto che utilizza gli stessi processori di NiFi, consentendo di creare gli stessi flow come in NiFi. La leggerezza dell'agente è raggiunta anche grazie al fatto che MiNiFy non ha un'interfaccia grafica per la configurazione dei flow. L'assenza di un'interfaccia grafica in MiNiFy significa che è necessario affrontare il problema della consegna del flow a minifi. Dato che MiNiFy è ampiamente usato nell'IOT, ci sono molti componenti e la consegna del flow agli esemplari finali di minifi deve essere automatizzata. Compito familiare, vero?

Per risolvere un compito del genere aiuta un altro sottoprogetto — MiNiFi C2 Server. Questo prodotto è progettato per essere il punto centrale nell'architettura di distribuzione delle configurazioni. Come configurare l'ambiente è descritto in questo articolo su Habr e le informazioni sono sufficienti per risolvere il compito assegnato. MiNiFi in combinazione con C2 server aggiorna automaticamente la propria configurazione. L'unico svantaggio di questo approccio è che è necessario creare modelli su C2 Server, un semplice commit nel registro non è sufficiente.

La variante descritta nell'articolo sopra è funzionale e non difficile da implementare, ma non bisogna dimenticare quanto segue:

1. In minifi non ci sono tutti i processori di nifi
2. Le versioni dei processori in Minifi sono indietro rispetto alle versioni dei processori in NiFi.

Al momento della scrittura della pubblicazione, l'ultima versione di NiFi è 1.9.2. La versione dei processori dell'ultima versione di MiNiFi è 1.7.0. È possibile aggiungere processori a MiNiFi, ma a causa delle discrepanze tra le versioni dei processori NiFi e MiNiFi, ciò potrebbe non funzionare.

NiFi CLI

A giudicare da descrizione dello strumento sul sito ufficiale, è uno strumento per automatizzare l'interazione tra NiFi e NiFi Registry nella consegna dei flow o nella gestione dei processi. Per iniziare a lavorare, è necessario scaricare questo strumento da qui.

Avviamo l'utilità

.\/bin\/cli.sh
           _     ___  _
 Apache   (_)  .' ..](_)   ,
 _ .--.   __  _| |_  __    )
[ `.-. | [  |'-| |-'[  |  \/  
|  | | |  | |  | |   | | '    '
[___||__][___][___] [___]',  ,'
                           `'
          CLI v1.9.2

Type 'help' per vedere un elenco dei comandi disponibili, usa il tab per completare automaticamente.

Per caricare il flow necessario dal registro, dobbiamo conoscere gli identificatori del bucket (bucket identifier) e del flow stesso (flow identifier). Questi dati possono essere ottenuti tramite cli o nell'interfaccia web di NiFi registry. Nell'interfaccia web appare così:

Con il CLI si fa così:

#> registry list-buckets -u http://nifi-registry:18080

#   Name             Id                                     Description
-   --------------   ------------------------------------   -----------
1   test_bucket   709d387a-9ce9-4535-8546-3621efe38e96   (empty)

#> registry list-flows -b 709d387a-9ce9-4535-8546-3621efe38e96 -u http://nifi-registry:18080

#   Name           Id                                     Description
-   ------------   ------------------------------------   -----------
1   test_flow   d27af00a-5b47-4910-89cd-9c664cd91e85

Avviamo l'importazione del gruppo di processi dal registro:

#> nifi pg-import -b 709d387a-9ce9-4535-8546-3621efe38e96 -f d27af00a-5b47-4910-89cd-9c664cd91e85 -fv 1 -u http://nifi:8080

7f522a13-016e-1000-e504-d5b15587f2f3

Un punto importante: come host su cui stiamo implementando il process group può essere specificato qualsiasi istanza di nifi.

Il process group è stato aggiunto con i processori bloccati, devono essere avviati.

#> nifi pg-start -pgid 7f522a13-016e-1000-e504-d5b15587f2f3 -u http://nifi:8080

Ottimo, i processori sono partiti. Tuttavia, secondo i requisiti del compito, abbiamo bisogno che le istanze NiFi inviino dati ad altre istanze. Supponiamo che per trasmettere i dati al server sia stato scelto il metodo Push. Per organizzare la trasmissione dei dati, è necessario attivare la trasmissione dei dati sul Remote Process Group (RPG) aggiunto, che è già incluso nel nostro flusso (Enable transmitting).

Nella documentazione nel CLI e in altre fonti non ho trovato un modo per abilitare la trasmissione dei dati. Se sapete come farlo, vi prego di scrivere un commento.

Dal momento che abbiamo bash e siamo pronti a perseverare, troviamo una soluzione! È possibile utilizzare l'API di NiFi per risolvere questo problema. Utilizzeremo il seguente metodo, l'ID viene preso dagli esempi sopra (nel nostro caso si tratta di 7f522a13-016e-1000-e504-d5b15587f2f3). Descrizione dei metodi dell'API di NiFi. qui.

Nel body è necessario inviare un JSON del seguente tipo:

{
    "revision": {
	    "clientId": "value",
	    "version": 0,
	    "lastModifier": "value"
	},
    "state": "value",
    "disconnectedNodeAcknowledged": true
}

I parametri da compilare per far funzionare tutto:
state — stato della trasmissione dei dati. È disponibile TRANSMITTING per abilitare la trasmissione dei dati, STOPPED per disabilitarla.
versione — versione del processore.

La versione di default sarà 0 alla creazione, ma possiamo ottenere questi parametri utilizzando il metodo.

Per gli amanti degli script bash, questo metodo potrebbe sembrare utile, ma a me risulta difficile: gli script bash non sono il mio forte. Il metodo successivo è più interessante e comodo, a mio avviso.

NiPyAPI

NiPyAPI è una libreria per il linguaggio Python per interagire con le istanze di NiFi. La pagina con la documentazione contiene le informazioni necessarie per lavorare con la libreria. Il rapido avvio è descritto nel progetto su github.

Il nostro script per implementare la configurazione è un programma in Python. Passiamo al codice.
Configuriamo i file di configurazione per il lavoro successivo. Abbiamo bisogno dei seguenti parametri:

nipyapi.config.nifi_config.host = 'http://nifi:8080/nifi-api' #percorso fino all'istanza nifi-api su cui implementiamo il process group
nipyapi.config.registry_config.host = 'http://nifi-registry:18080/nifi-registry-api' #percorso fino all'istanza nifi-registry-api registry
nipyapi.config.registry_name = 'MyBeutifulRegistry' #nome del registry, come verrà chiamato nell'istanza nifi
nipyapi.config.bucket_name = 'BucketName' #nome del bucket, da cui estraiamo il flusso
nipyapi.config.flow_name = 'FlowName' #nome del flusso, che estraiamo

Continuerò a inserire i nomi dei metodi di questa libreria che sono descritti qui.

Colleghiamo il registry all'istanza di nifi utilizzando

nipyapi.versioning.create_registry_client

In questo passaggio puoi anche aggiungere un controllo per verificare se il registry è già stato aggiunto all'istanza, per fare ciò puoi utilizzare il metodo

nipyapi.versioning.list_registry_clients

Troviamo il bucket per cercare ulteriormente il flow nel cestino

nipyapi.versioning.get_registry_bucket

Dopo aver trovato il bucket, cerchiamo il flow

nipyapi.versioning.get_flow_in_bucket

È importante capire se questo process group è già stato aggiunto. I process group si posizionano secondo le coordinate e potrebbe verificarsi una situazione in cui un secondo componente si sovrappone a uno già presente. Ho verificato, può succedere 🙂 Per ottenere tutti i process group aggiunti utilizziamo il metodo

nipyapi.canvas.list_all_process_groups

e poi possiamo cercare, ad esempio, per nome.

Non descriverò il processo di aggiornamento del template, dirò solo che se nella nuova versione del template vengono aggiunti i processori, non ci sono problemi con la presenza di messaggi nelle code. Tuttavia, se i processori vengono rimossi, potrebbero sorgere problemi (nifi non consente di rimuovere un processore se ci sono messaggi in coda davanti a esso). Se sei interessato a sapere come ho risolto questo problema, per favore scrivimi, ne discuteremo. I contatti sono alla fine dell'articolo. Passiamo al passaggio di aggiunta del process group.

Durante il debug dello script ho riscontrato un'peculiarità: non sempre viene recuperata l'ultima versione del flow, quindi consiglio prima di chiarire questa versione:

nipyapi.versioning.get_latest_flow_ver

Deployiamo il process group:

nipyapi.versioning.deploy_flow_version

Avviamo i processori:

Nel blocco riguardante il CLI era stato menzionato che nella remote process group la trasmissione dei dati non si attiva automaticamente? Durante l'implementazione dello script ho riscontrato anche io questo problema. In quel momento, non riuscivo a avviare la trasmissione dei dati tramite l'API e ho deciso di scrivere allo sviluppatore della libreria NiPyAPI per chiedere consiglio/aiuto. Lo sviluppatore mi ha risposto, abbiamo discusso del problema e ha scritto che aveva bisogno di tempo “per controllare qualcosa”. E così, dopo un paio di giorni ricevo una email, in cui c'è una funzione in Python che risolve il mio problema di avvio!!! A quel punto la versione di NiPyAPI era 0.13.3 e ovviamente non c'era nulla di tutto ciò. Ma nella versione 0.14.0, che è stata rilasciata di recente, questa funzione è già inclusa nella libreria. Ecco a voi,

nipyapi.canvas.set_remote_process_group_transmission

nipyapi.canvas.set_remote_process_group_transmission

Quindi, abbiamo collegato il registry con la libreria NiPyAPI, implementato il flow e persino avviato i processori e il trasferimento dei dati. Ora possiamo affinare il codice, aggiungere vari controlli, logging e tutto il resto. Ma questa è già un'altra storia.

Tra le varie opzioni di automazione che ho esaminato, l'ultima mi è sembrata la più funzionale. In primo luogo, si tratta comunque di codice Python, che può incorporare codice ausiliario e sfruttare tutti i vantaggi del linguaggio di programmazione. In secondo luogo, il progetto NiPyAPI è in continua evoluzione e in caso di problemi è possibile contattare lo sviluppatore. In terzo luogo, NiPyAPI è un strumento più flessibile per interagire con NiFi nella risoluzione di compiti complessi. Ad esempio, per determinare se le code dei messaggi siano vuote nel flow e se sia possibile aggiornare il process group.

Questo è tutto. Ho descritto 3 approcci all'automazione della consegna del flow in NiFi, le insidie con cui un sviluppatore può imbattersi e fornito codice funzionante per automatizzare la consegna. Se ti interessa questo argomento quanto interessa a me — scrivimi!

Fonte: habr.com