Architecture de NiFi expliquée avec des exemples pratiques

Introduction

Apache NiFi est un outil puissant d'automatisation de flux de données, mais son architecture peut sembler déroutante au premier abord. Ce guide explore l'architecture de NiFi à l'aide d'exemples pratiques, aidant les opérateurs à passer d'un problème observé à un résultat vérifié. Que vous soyez développeur, consultant DevOps ou membre d'une startup technique, vous apprendrez à connecter les composants NiFi, à concevoir des flux de données fiables et à les exploiter en toute sécurité.

Nous mettons l'accent sur la sécurité opérationnelle : observez avant de modifier, limitez le rayon d'impact, utilisez des espaces réservés plutôt que des secrets, vérifiez les résultats et documentez les chemins de récupération. Chaque recommandation est délimitée par version, observable et réversible lorsque la technologie le permet.

Inventaire de la version et de l'environnement

Avant toute modification d'un déploiement NiFi, vous devez savoir exactement avec quoi vous travaillez. Commencez par identifier la version installée et la topologie de déploiement. Cela évite d'appliquer des conseils destinés à une version ou une configuration différente.

Observation en lecture seule

Pour vérifier la version de NiFi depuis la ligne de commande, utilisez le script nifi.sh avec l'argument status (sur une installation standard) :

/path/to/nifi/bin/nifi.sh status

La sortie attendue sur NiFi 1.x comprend une ligne semblable à :

Java home: /usr/lib/jvm/java-11-openjdk-amd64
NiFi home: /opt/nifi

Bootstrap File: /opt/nifi/conf/bootstrap.conf

2023-08-01 12:00:00,000 INFO [main] org.apache.nifi.bootstrap.Command - Apache NiFi is currently running, PID 12345

Pour NiFi 2.x, le script a pu changer ; consultez toujours la documentation officielle de votre version. Pour les déploiements conteneurisés, utilisez :

docker exec <container_name> /opt/nifi/bin/nifi.sh status

Prérequis

Version Java compatible avec votre version de NiFi (NiFi 1.20+ requiert Java 11 ou 17 ; NiFi 2.0 requiert Java 21).
Accès au répertoire d'installation de NiFi ou au conteneur.
Autorisations de lecture sur les fichiers de configuration si vous les inspectez.

Rayon d'impact et changement

Si vous déterminez que la version installée est la 1.20.0 et que vous souhaitiez utiliser des fonctionnalités de la 2.0, le plus petit changement justifié est de procéder à une mise à niveau vers une version prise en charge. Avant cela, sauvegardez votre répertoire conf et les définitions de flux. Le rayon d'impact inclut tous les processeurs et services de contrôleur, car ils peuvent présenter des changements de compatibilité.

Vérification

Après une mise à niveau, vérifiez à nouveau la version avec nifi.sh status ou l'API REST :

curl -k https://localhost:8443/nifi-api/system-diagnostics

La sortie attendue inclut "version":"2.0.0" dans la réponse JSON. Si vous voyez une version différente, revenez en arrière à l'aide de votre sauvegarde et consultez les notes de migration.

Chemin de configuration sûr

Une configuration sûre dans NiFi consiste à séparer l'observation de l'intervention. De nombreux incidents en production commencent par un changement bien intentionné qui n'a pas été délimité ni rendu réversible. Voici un chemin pratique pour modifier une propriété de processeur.

Exemple : Modifier la période d'ordonnancement d'un processeur

Supposons que vous deviez ajuster la planification d'un processeur GetFile de 60 secondes à 10 secondes pour réduire la latence.

Observez l'état actuel via l'API REST de NiFi (remplacez <nifi-host> et <processor-id>) :

curl -k -u admin:password https://<nifi-host>:8443/nifi-api/processors/<processor-id>

Cherchez le champ schedulingPeriod dans la réponse JSON. Il peut indiquer "schedulingPeriod":"60 sec".

Vérifiez les prérequis : Vous avez besoin de l'ID du processeur. Dans l'interface utilisateur de NiFi, faites un clic droit sur le processeur et sélectionnez « View Configuration » pour trouver l'ID dans l'URL, ou utilisez l'API pour lister les processeurs d'un groupe de processus.

Effectuez le plus petit changement : Utilisez une requête PUT pour mettre à jour uniquement la période d'ordonnancement. Construisez une charge utile JSON avec la révision actuelle et la nouvelle valeur :

{
  "revision": { "clientId": "my-client", "version": 3 },
  "component": {
    "id": "<processor-id>",
    "config": {
      "schedulingPeriod": "10 sec"
    }
  }
}

Puis soumettez :

curl -k -X PUT -u admin:password -H "Content-Type: application/json" -d @update.json https://<nifi-host>:8443/nifi-api/processors/<processor-id>

Rayon d'impact : Ce changement n'affecte que la fréquence d'exécution du processeur spécifié. Il ne modifie pas le contenu des données ni les relations de flux. Cependant, une période plus courte peut augmenter la charge du système.

Vérifiez : Récupérez à nouveau le processeur et confirmez "schedulingPeriod":"10 sec". Surveillez également la taille des files d'attente pour vous assurer que le processeur suit le rythme.

Récupération : Si le changement cause des problèmes (par exemple, une utilisation excessive du CPU), rétablissez la période d'ordonnancement à sa valeur d'origine en utilisant la même API avec la bonne révision.

Note de sécurité : Ne mettez jamais de véritables identifiants dans les scripts ou les exemples. Utilisez des variables d'environnement ou des coffres sécurisés. Pour l'exemple ci-dessus, remplacez admin:password par une référence à un identifiant sécurisé.

Vérification et diagnostics

La vérification consiste à prouver que le système se comporte comme prévu après un changement ou lors des opérations de routine. NiFi fournit plusieurs outils de diagnostic.

Utilisation de nifi.sh diagnostics

La commande diagnostics génère un rapport d'informations sur le système et les flux. Exécutez :

/path/to/nifi/bin/nifi.sh diagnostics <output-directory>

Cela crée un fichier zip contenant les journaux, la configuration et des vidages de threads. Consultez system-diagnostics.txt pour la mémoire et la charge du processeur :

Total Memory: 16.0 GB
Used Memory: 3.2 GB
Free Memory: 12.8 GB

Vérification de l'état d'un processeur via l'API

Pour vérifier qu'un processeur est en cours d'exécution et a traité des données, utilisez :

curl -k -u admin:password https://<nifi-host>:8443/nifi-api/processors/<processor-id>/status

Recherchez des champs comme "runStatus":"Running" et des compteurs agrégés de fichiers de flux traités. Si runStatus est "Stopped" de manière inattendue, c'est un signal de défaillance.

Exemple de dépannage

Supposons qu'un processeur PutDatabaseRecord échoue. Vérifiez le bulletin du processeur :

curl -k -u admin:password https://<nifi-host>:8443/nifi-api/processors/<processor-id>/bulletins

Si vous voyez une erreur comme "Cannot get connection for url jdbc:...", vérifiez la chaîne de connexion à la base de données et les identifiants dans le service de contrôleur DBCPConnectionPool. Effectuez un changement à la fois et redémarrez le processeur.

Liste de vérification

Comparez les métriques avant et après (tailles de files d'attente, nombre de fichiers de flux, taux d'erreurs).
Confirmez que les fichiers de sortie attendus existent ou que les systèmes cibles ont reçu les données.
Pour l'intégrité des données, utilisez des sommes de contrôle ou des comptages de lignes.

Modes de défaillance et récupération

Comprendre les modes de défaillance courants vous aide à préparer des procédures de récupération. Voici trois scénarios pratiques.

1. Espace disque insuffisant

Symptôme : Les processeurs échouent avec "No space left on device" dans les journaux, ou les fichiers de flux s'accumulent dans les files d'attente.

Diagnostic : Vérifiez l'utilisation du disque :

df -h /path/to/nifi/repositories

Si l'utilisation est proche de 100 %, identifiez les grands dépôts de contenu ou de fichiers de flux.

Récupération (choisissez-en une, la plus petite d'abord) :

Augmentez l'espace disque ou ajoutez un nouvel emplacement de dépôt de contenu dans nifi.properties (nifi.content.repository.directory.*).
Nettoyez les anciens événements de provenance via l'interface utilisateur (menu Global -> Data Provenance -> supprimer les anciens événements).
Si c'est sûr, supprimez les anciens fichiers journaux.

Vérification : Après avoir libéré de l'espace, exécutez à nouveau df -h pour confirmer une utilisation inférieure à 80 %. Redémarrez les processeurs affectés.

2. Processeur bloqué en état « Running » sans progression

Symptôme : Le processeur indique « Running » mais aucun fichier de flux ne passe, et les bulletins ne montrent aucune erreur.

Diagnostic : Soupçonnez un blocage de thread ou une tâche de longue durée. Utilisez nifi.sh diagnostics pour obtenir des vidages de threads. Cherchez des threads bloqués sur le même verrou.

Récupération : Arrêtez le processeur. S'il ne s'arrête pas, redémarrez l'instance NiFi. Après le redémarrage, si le problème se reproduit, inspectez la configuration du processeur et ses dépendances (par exemple, un service de contrôleur qui n'est pas activé).

Vérification : Activez le processeur et surveillez la taille de la file d'attente ; elle devrait commencer à se vider.

3. Fichier de flux bloqué dans une boucle

Symptôme : Un processeur comme RouteOnAttribute renvoie un fichier vers lui-même, provoquant des boucles infinies.

Diagnostic : Inspectez les connexions de flux de données et les règles de routage. La lignée d'un fichier montre des passages répétés par le même processeur.

Récupération : Modifiez la logique de routage pour envoyer les échecs vers un processeur LogAttribute ou une file d'attente de lettres mortes au lieu de boucler en arrière. Testez le changement sur un seul fichier de flux.

Vérification : Après le changement, aucun nouveau fichier n'entre dans la boucle ; les fichiers bloqués existants sont déplacés manuellement ou supprimés.

Principes généraux de récupération

Capturez toujours l'état avant d'intervenir : exportez la définition de flux (clic droit sur le canevas -> Download flow) comme sauvegarde.
Documentez la commande exacte ou l'appel API utilisé pour la récupération.
Testez les procédures de récupération dans un environnement de préproduction lorsque c'est possible.

Liste de contrôle des opérations

Utilisez cette liste de contrôle pour les opérations NiFi de routine afin de prévenir les incidents et d'assurer une récupération rapide.

Vérifications quotidiennes (automatisées par script)

Créez un script shell qui s'exécute chaque matin et alerte en cas d'anomalies :

#!/bin/bash
# Vérifier que NiFi est en cours d'exécution
if /path/to/nifi/bin/nifi.sh status | grep -q "Apache NiFi is currently running"; then
  echo "NiFi is running"
else
  echo "NiFi is not running - alert!"
fi

# Vérifier l'utilisation du disque (avertissement à 80%)
USAGE=$(df -h /opt/nifi | awk 'NR==2 {print $5}' | tr -d '%')
if [ "$USAGE" -gt 80 ]; then
  echo "Disk usage is $USAGE% - alert!"
fi

Planifiez avec cron :

0 7 * * * /usr/local/bin/nifi_daily_check.sh

Santé des processeurs et des services de contrôleur

Listez tous les processeurs avec leur état d'exécution :

curl -k -u admin:password https://<nifi-host>:8443/nifi-api/processors

Recherchez tout processeur avec "runStatus":"Invalid" ou "Stopped" qui devrait être en cours d'exécution.

Assurez-vous que tous les services de contrôleur requis sont activés :

curl -k -u admin:password https://<nifi-host>:8443/nifi-api/controller-services

Filtrez pour "state":"ENABLED" là où c'est attendu.

Sauvegarde et restauration

Avant tout changement de configuration, exportez le flux :

Dans l'interface utilisateur de NiFi, allez au groupe de processus racine, faites un clic droit et sélectionnez « Download flow definition » (JSON). Stockez-le avec un horodatage :

mv downloaded_flow.json flow_backup_$(date +%Y%m%d).json

Pour restaurer, importez le JSON via l'interface utilisateur ou l'API.

Considérations spécifiques à la version

NiFi 2.x a introduit des changements dans nipyapi et une nouvelle intégration du registre. Vérifiez toujours la documentation de votre version. Par exemple, le point de terminaison de l'API pour process-groups est passé de /nifi-api/process-groups/{id} à l'inclusion de paramètres de requête supplémentaires.

Liste de contrôle de sécurité

Utilisez HTTPS avec des certificats valides ; ne désactivez jamais la validation des certificats en production.
Utilisez la gestion des utilisateurs intégrée de NiFi ou intégrez-la avec LDAP/OpenID Connect.
Faites tourner les secrets (mots de passe de keystore, identifiants de base de données) périodiquement et mettez-les à jour dans nifi.properties ou les variables sécurisées.
Auditez les accès via les événements de provenance.

Conclusion

L'architecture de NiFi expliquée avec des exemples pratiques ne devient utile que si chaque recommandation est délimitée par version, observable et réversible lorsque la technologie le permet. Copier une commande sans vérifier les prérequis et la sortie attendue n'est pas une procédure d'exploitation.

Comme prochaine étape, choisissez une vérification à faible risque pour votre configuration NiFi : enregistrez l'état actuel, exécutez une vérification documentée (par exemple, nifi.sh status ou un appel API d'état de processeur), comparez le résultat avec le signal attendu et passez en revue les dépendances telles que Kafka, HDFS et Apache Spark si elles font partie de votre flux.

Un flux de travail technique fiable rend les défaillances visibles, protège les valeurs sensibles, limite les changements à la ressource prévue et définit la vérification de récupération avant qu'un incident ne force la décision. Commencez petit, observez et itérez.