¡Hola a todos!
La tarea es la siguiente: hay un flow, representado en la imagen de arriba, que debe desplegarse en N servidores con . El flow es de prueba: se genera un archivo y se envía a otra instancia de NiFi. La transferencia de datos se realiza mediante el protocolo NiFi Site to Site.
NiFi Site to Site (S2S) es una forma segura y fácil de configurar para transferir datos entre instancias de NiFi. Cómo funciona S2S, véase en y es importante no olvidar configurar la instancia de NiFi para permitir S2S, véase .
En los casos en que se trata de la transferencia de datos mediante S2S, una instancia se llama cliente y la otra servidor. El cliente envía los datos y el servidor los recibe. Hay dos maneras de configurar la transferencia de datos entre ellos:
- Push. Desde la instancia cliente, los datos se envían mediante Remote Process Group (RPG). En la instancia servidor, los datos se reciben a través de Input Port
- Pull. El servidor recibe los datos mediante RPG, el cliente envía mediante Output port.
El flow para el despliegue lo guardamos en Apache Registry.
Apache NiFi Registry es un subproyecto de Apache NiFi, que actúa como una herramienta para almacenar flows y gestionar versiones. Una especie de GIT. La información sobre la instalación, configuración y trabajo con el registry se puede encontrar en . Los flows para el almacenamiento se combinan en un process group y se almacenan en el registry de esta manera. Más adelante en el artículo volveremos a esto.
Al inicio, cuando N es un número pequeño, el flow se entrega y actualiza manualmente en un tiempo razonable.
Pero con el aumento de N, los problemas aumentan:
- la actualización del flow lleva más tiempo. Hay que acceder a todos los servidores
- aparecen errores en la actualización de las plantillas. Aquí se actualizó, pero aquí se olvidó
- errores humanos al realizar un gran número de operaciones similares
Todo esto nos lleva a la conclusión de que es necesario automatizar el proceso. He probado las siguientes formas de resolver esta tarea:
- Usar MiNiFi en lugar de NiFi
- NiFi CLI
- NiPyAPI
Uso de MiNiFi
— subproyecto Apache NiFi. MiNiFy — un agente compacto que utiliza los mismos procesadores que NiFi, permitiendo crear los mismos flujos que en NiFi. La ligereza del agente se logra en parte por no tener una interfaz gráfica para la configuración de flujos. La ausencia de una interfaz gráfica en MiNiFy significa que es necesario resolver el problema de entregar flujos a Minifi. Dado que MiNiFy se utiliza activamente en IoT, hay muchos componentes y el proceso de entrega de flujos a las instancias finales de Minifi debe ser automatizado. Es una tarea familiar, ¿verdad?
Resolver tal tarea será posible con otro subproyecto — MiNiFi C2 Server. Este producto está diseñado para ser el punto central en la arquitectura de despliegue de configuraciones. Cómo configurar el entorno está descrito en en Habr y hay suficiente información para resolver la tarea planteada. MiNiFi, en conjunto con el servidor C2, actualiza automáticamente su configuración. El único inconveniente de este enfoque es que es necesario crear plantillas en el servidor C2, un simple commit en el registro no es suficiente.
La opción descrita en el artículo anterior es funcional y no complicada de implementar, pero no debemos olvidar lo siguiente:
- En MiNiFi no están todos los procesadores de NiFi.
- Las versiones de los procesadores en MiNiFi están desactualizadas en comparación con las versiones de los procesadores en NiFi.
En el momento de escribir esta publicación, la última versión de NiFi es 1.9.2. La versión de los procesadores de la última versión de MiNiFi es 1.7.0. Se pueden añadir procesadores a MiNiFi, pero debido a la discrepancia de versiones entre los procesadores de NiFi y MiNiFi, esto puede no funcionar.
NiFi CLI
Según el de la herramienta en el sitio web oficial, esta es una herramienta para automatizar la interacción entre NiFi y NiFi Registry en el ámbito de la entrega de flujos o gestión de procesos. Para empezar a trabajar, esta herramienta debe ser descargada .
Ejecutamos la utilidad
.\/bin\/cli.sh
_ ___ _
Apache (_) .' ..](_) ,
_ .--. __ _| |_ __ )
[ `.-. | [ |'-| |-'[ | \/
| | | | | | | | | | ' '
[___||__][___][___] [___]', ,'
`'
CLI v1.9.2
Type 'help' to see a list of available commands, use tab to auto-complete.
Para cargar el flujo necesario desde el registro, necesitamos conocer los identificadores del bucket (bucket identifier) y del flujo mismo (flow identifier). Estos datos se pueden obtener bien a través del CLI o en la interfaz web de NiFi Registry. En la interfaz web se ve así:
Con el CLI se hace así:
#> 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
Ejecutamos la importación del grupo de procesos desde el 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: como host al que aplicamos el grupo de procesos puede ser cualquier instancia de NiFi.
El grupo de procesos se ha agregado con procesadores detenidos, necesitan ser iniciados.
#> nifi pg-start -pgid 7f522a13-016e-1000-e504-d5b15587f2f3 -u http://nifi:8080
Excelente, los procesadores han arrancado. Sin embargo, según las condiciones del problema, necesitamos que las instancias de NiFi envíen datos a otras instancias. Supongamos que para transmitir los datos al servidor elegimos el método Push. Para organizar la transmisión de datos, debemos habilitar la transmisión de datos en el Remote Process Group (RPG) agregado, que ya está incluido en nuestro flujo.
No he encontrado en la documentación de CLI ni en otras fuentes una forma de habilitar la transmisión de datos. Si sabes cómo hacerlo, por favor, escribe en los comentarios.
Dado que tenemos bash y estamos listos para continuar, ¡encontremos la solución! Podemos usar la API de NiFi para resolver este problema. Utilizaremos el siguiente método, el ID se toma de los ejemplos anteriores (en nuestro caso es 7f522a13-016e-1000-e504-d5b15587f2f3). Descripción de los métodos de la API de NiFi. .
En el cuerpo se debe enviar un JSON de la siguiente forma:
{
"revision": {
"clientId": "valor",
"version": 0,
"lastModifier": "valor"
},
"state": "valor",
"disconnectedNodeAcknowledged": true
}
Los parámetros que deben completarse para que funcione:
state — estado de transmisión de datos. Disponible TRANSMITTING para habilitar la transmisión de datos, STOPPED para deshabilitarla.
version — versión del procesador.
La versión por defecto será 0 al crearla, pero estos parámetros se pueden obtener usando el método.
Para aquellos amantes de los scripts bash, este método puede parecer adecuado, pero me resulta difícil, ya que los scripts bash no son lo mío. El siguiente método es más interesante y conveniente a mi parecer.
NiPyAPI
NiPyAPI — una biblioteca para el lenguaje Python para interactuar con instancias de NiFi. contiene la información necesaria para trabajar con la biblioteca. El inicio rápido se describe en en github.
Nuestro script para implementar la configuración es un programa en lenguaje Python. Pasemos a la codificación.
Configuramos los archivos de configuración para el trabajo posterior. Necesitaremos los siguientes parámetros:
nipyapi.config.nifi_config.host = 'http://nifi:8080/nifi-api' #ruta hasta la API de nifi en la que desplegamos el grupo de procesos
nipyapi.config.registry_config.host = 'http://nifi-registry:18080/nifi-registry-api' #ruta hasta la API de nifi-registry
nipyapi.config.registry_name = 'MyBeutifulRegistry' #nombre del registro, como se llamará en la instancia de nifi
nipyapi.config.bucket_name = 'BucketName' #nombre del bucket del que extraemos el flujo
nipyapi.config.flow_name = 'FlowName' #nombre del flujo que extraemos.
A continuación, iré insertando los nombres de los métodos de esta biblioteca, que se describen .
Conectamos el registro a la instancia de nifi mediante
nipyapi.versioning.create_registry_clientEn este paso, también se puede añadir una verificación para asegurarse de que el registro ya se haya añadido a la instancia, para lo cual se puede utilizar el método
nipyapi.versioning.list_registry_clientsBuscamos el bucket para la búsqueda posterior del flujo en la cesta
nipyapi.versioning.get_registry_bucketCon el bucket encontrado, buscamos el flujo
nipyapi.versioning.get_flow_in_bucketLuego es importante entender si este grupo de procesos ya ha sido añadido. El grupo de procesos se sitúa en coordenadas y puede haber situaciones donde un componente se superponga a otro. Lo he comprobado, eso puede ocurrir 🙂. Para obtener todos los grupos de procesos añadidos, utilizamos el método
nipyapi.canvas.list_all_process_groupsy luego podemos buscar, por ejemplo, por nombre.
No voy a describir el proceso de actualización de la plantilla, solo diré que si en la nueva versión de la plantilla se añaden procesadores, no hay problemas con la existencia de mensajes en las colas. Pero si se eliminan procesadores, pueden surgir problemas (nifi no permite borrar un procesador si hay una cola de mensajes acumulada frente a él). Si te interesa cómo resolví este problema, por favor, escríbeme, lo discutiremos. Mis contactos están al final del artículo. Pasemos al paso de añadir el grupo de procesos.
Al depurar el script, me encontré con la peculiaridad de que no siempre se retrieve la última versión del flujo, por lo que recomiendo primero verificar esta versión:
nipyapi.versioning.get_latest_flow_verDesplegamos el grupo de procesos:
nipyapi.versioning.deploy_flow_versionIniciamos los procesadores:
En el bloque sobre CLI se mencionó que en el grupo de procesos remoto no se incluye automáticamente la transmisión de datos. Al implementar el script, también me encontré con este problema. En ese momento, no pude iniciar la transmisión de datos a través de la API y decidí escribirle al desarrollador de la biblioteca NiPyAPI para pedir consejo/ayuda. El desarrollador me respondió, discutimos el problema y él escribió que necesitaba tiempo para 'verificar algunas cosas'. Y así, después de un par de días, recibí un correo con una función en Python que resolvía mi problema de inicio. En ese momento, la versión de NiPyAPI era 0.13.3 y, por supuesto, no contenía nada de esto. Pero en la versión 0.14.0, que salió muy recientemente, esta función ya formaba parte de la biblioteca. ¡Bienvenidos!nipyapi.canvas.set_remote_process_group_transmission
nipyapi.canvas.set_remote_process_group_transmissionAsí que, con la biblioteca NiPyAPI, conectamos el registro, implementamos el flujo e incluso iniciamos los procesadores y la transferencia de datos. A partir de aquí, se puede refinar el código, añadir diversas comprobaciones, registro y todo eso. Pero eso ya es una historia completamente diferente.
De las opciones de automatización que he considerado, la última me parece la más eficaz. En primer lugar, sigue siendo código en Python, al que se le puede integrar código auxiliar y aprovechar todas las ventajas del lenguaje de programación. En segundo lugar, el proyecto NiPyAPI está en desarrollo activo y, en caso de problemas, se puede contactar al desarrollador. En tercer lugar, NiPyAPI es una herramienta más flexible para interactuar con NiFi en la resolución de tareas complejas. Por ejemplo, al determinar si las colas de mensajes están vacías en el flujo y si se puede actualizar el grupo de procesos.
Eso es todo. He descrito 3 enfoques para automatizar la entrega de flujos en NiFi, los escollos que pueden encontrar los desarrolladores y he proporcionado código funcional para la automatización de la entrega. Si este tema también te interesa, como a mí —
Fuente: habr.com
