Ciao a tutti!

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

NiFi Site to Site (S2S) è un modo sicuro e facilmente configurabile per trasferire dati tra istanze di NiFi. Come funziona S2S, consulta documentazione ed è importante non dimenticare di configurare l'istanza di NiFi per consentire S2S, consulta qui.

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

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

Il flow da distribuire è memorizzato in Apache Registry.

Apache NiFi Registry è un sottoprogetto di Apache NiFi, rappresenta uno strumento per memorizzare flow e gestire versioni. Una sorta di GIT. Le informazioni su installazione, configurazione e utilizzo del registry possono essere trovate in documentazione ufficiale. Flow per la memorizzazione si unisce in una group di processi e viene registrato in questo modo nel registro. Torneremo a questo più avanti nell'articolo.

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

Ma con l'aumento di N, i problemi aumentano:

1. l'aggiornamento del flow richiede più tempo. Dobbiamo accedere a tutti i server
2. si verificano 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 metodi per risolvere questo problema:

1. Utilizzare MiNiFi al posto di NiFi
2. NiFi CLI
3. NiPyAPI

Utilizzo di MiNiFi

Apache MiNiFi — sottoprogetto di Apache NiFi. MiNiFy è un agente compatto che utilizza gli stessi processori di NiFi, consentendo di creare gli stessi flow di NiFi. La leggerezza dell'agente è dovuta anche al fatto che MiNiFy non dispone di un'interfaccia grafica per la configurazione dei flow. L'assenza dell'interfaccia grafica in MiNiFy implica la necessità di affrontare il problema della consegna del flow a minifi. Poiché MiNiFy è ampiamente utilizzato nell'IOT, ci sono molti componenti e la consegna del flow ai singoli istanti di minifi deve essere automatizzata. Un compito familiare, vero?

Per affrontare tale problematica, ci aiuterà un altro sottoprogetto: MiNiFi C2 Server. Questo prodotto è progettato per fungere da punto centrale nell'architettura della distribuzione delle configurazioni. Come configurare l'ambiente è descritto in questo articolo su Habr e ci sono sufficienti informazioni per risolvere il problema. MiNiFi, in combinazione con il server C2, aggiorna automaticamente la propria configurazione. L'unico svantaggio di questo approccio è che è necessario creare modelli sul server C2; una semplice commit nel registry non è sufficiente.

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

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

Al momento della pubblicazione, l'ultima versione di NiFi è 1.9.2. La versione dei processori dell'ultima versione di MiNiFi è 1.7.0. I processori possono essere aggiunti a MiNiFi, ma a causa delle differenze di versione tra i processori di NiFi e MiNiFi, potrebbero non funzionare.

NiFi CLI

Secondo descrizione strumento sul sito ufficiale, è uno strumento per automatizzare l'interazione tra NiFi e NiFi Registry nell'ambito della consegna dei flow o della gestione dei processi. Per iniziare a utilizzare questo strumento, è necessario scaricarlo da qui.

Avviamo l'utilità

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

Type 'help' to see a list of available commands, use tab to auto-complete.

Per caricare il flow necessario dal registry, 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 la 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 process group dal registry:

#> 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 aspetto importante è che come host su cui distribuiamo il process group può essere indicato qualsiasi istanza di nifi.

Il process group è stato aggiunto con i processori fermati, è necessario avviarli.

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

Ottimo, i processori sono partiti. Tuttavia, secondo i requisiti, abbiamo bisogno che le istanze di NiFi inviino dati ad altre istanze. Supponiamo che per la trasmissione dei dati al server sia stata scelta la modalità Push. Per organizzare la trasmissione dei dati, è necessario abilitare la trasmissione dei dati nel Remote Process Group (RPG) aggiunto, che è già incluso nel nostro flow (Enable transmitting).

Nella documentazione CLI e in altre fonti non ho trovato un modo per abilitare la trasmissione dei dati. Se sai come farlo, ti prego di scriverlo nei commenti.

Poiché abbiamo bash e siamo pronti a procedere — troviamo una soluzione! Possiamo utilizzare l'API NiFi per risolvere questo problema. Utilizzeremo il seguente metodo, ID preso dagli esempi precedenti (nel nostro caso è 7f522a13-016e-1000-e504-d5b15587f2f3). Descrizione dei metodi dell'API NiFi. qui.

Nel body dobbiamo inviare JSON della seguente forma:

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

I parametri da compilare affinché funzioni:
state — stato di trasmissione dei dati. Disponibile TRANSMITTING per attivare la trasmissione dei dati, STOPPED per disattivarla
version — versione del processore

la versione predefinita sarà 0 al momento della creazione, ma questi parametri possono essere ottenuti utilizzando il metodo

Per gli amanti degli script bash, questo metodo potrebbe sembrare utile, ma per me è un po' difficile — gli script bash non sono esattamente la mia passione. Il prossimo metodo è più interessante e comodo, secondo me.

NiPyAPI

NiPyAPI — una libreria per il linguaggio Python per interagire con le istanze di NiFi. La pagina della documentazione contiene le informazioni necessarie per lavorare con la libreria. Il quick start è descritto in progetto su github.

Il nostro script per il rollout della configurazione è un programma scritto in Python. Passiamo al coding.
Configuriamo i file di configurazione per il lavoro futuro. Avremo bisogno dei seguenti parametri:

nipyapi.config.nifi_config.host = 'http://nifi:8080/nifi-api' #percorso fino all'istanza nifi-api su cui distribuiamo il gruppo di processi
nipyapi.config.registry_config.host = 'http://nifi-registry:18080/nifi-registry-api' #percorso fino all'istanza nifi-registry-api
nipyapi.config.registry_name = 'MyBeutifulRegistry' #nome del registry, come verrà chiamato nell'istanza di nifi
nipyapi.config.bucket_name = 'BucketName' #nome del bucket da cui preleviamo il flusso
nipyapi.config.flow_name = 'FlowName' #nome del flusso che stiamo prelevando

A questo punto inizierò 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 questa fase è possibile aggiungere anche un controllo per verificare se il registry è già stato aggiunto all'istanza, per questo possiamo utilizzare il metodo

nipyapi.versioning.list_registry_clients

Troviamo il bucket per ulteriori ricerche del flow nel cestino

nipyapi.versioning.get_registry_bucket

In base al bucket trovato cerchiamo il flow

nipyapi.versioning.get_flow_in_bucket

È inoltre importante capire se questo process group è già stato aggiunto. Il process group viene posizionato secondo coordinate e può verificarsi una situazione in cui un componente si sovrappone a un altro. Ho verificato, può succedere 🙂 Per ottenere tutti i process groups 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 processori, non ci sono problemi con la presenza di messaggi nelle code. Tuttavia, se i processori vengono rimossi, possono sorgere problemi (nifi non consente di rimuovere un processore se davanti a esso si è accumulata una coda di messaggi). Se sei interessato a come ho risolto questo problema, per favore scrivimi, ne discuteremo. I contatti sono alla fine dell'articolo. Passiamo quindi al passaggio di aggiunta del process group.

Durante il debug dello script ho riscontrato una particolare caratteristica: l'ultima versione del flow non sempre viene caricata, quindi ti consiglio di chiarire prima questa versione:

nipyapi.versioning.get_latest_flow_ver

Deployiamo il process group:

nipyapi.versioning.deploy_flow_version

Avviamo i processori:

nipyapi.canvas.schedule_process_group

Nella sezione sul CLI, era stato scritto che la trasmissione dei dati non è inclusa automaticamente nel remote process group? Anche io ho incontrato questo problema mentre implementavo lo script. In quel momento, non sono riuscito a avviare la trasmissione dei dati tramite API e ho deciso di contattare lo sviluppatore della libreria NiPyAPI per chiedere consiglio/aiuto. Lo sviluppatore mi ha risposto, abbiamo discusso del problema e lui ha scritto che avrebbe bisogno di tempo per "controllare alcune cose". E così, dopo un paio di giorni, ricevo un'email in cui è presente una funzione in Python che risolve il mio problema di avvio!!! A quel tempo, la versione di NiPyAPI era 0.13.3 e ovviamente non conteneva nulla del genere. Ma nella versione 0.14.0, che è stata rilasciata di recente, questa funzione è stata già inclusa nella libreria. Ecco a voi,

nipyapi.canvas.set_remote_process_group_transmission

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

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

Questo è tutto. Ho descritto 3 approcci per l'automazione della distribuzione del flow in NiFi, le insidie con cui può confrontarsi uno sviluppatore e ho fornito codice funzionante per l'automazione della distribuzione. Se sei interessato a questo tema come lo sono io — scrivimi!

Fonte: habr.com