Tutoriel Node-RED InfluxDB : stocker et interroger des données MQTT

Node-RED et InfluxDB forment ensemble un pipeline de télémétrie compact mais complet : les messages MQTT arrivent, Node-RED les met en forme et InfluxDB les stocke sous forme de séries temporelles que vous pouvez interroger et représenter graphiquement. Dans ce tutoriel Node-RED InfluxDB, vous allez lancer InfluxDB 2 sous Docker, écrire des mesures MQTT avec des fields et des tags bien choisis, puis les relire en Flux pour tracer un graphique. Chaque étape utilise un broker gratuit : vous pouvez tester toute la chaîne en moins d'une demi-heure.

Architecture : MQTT en entrée, séries temporelles en sortie

Le pipeline comporte trois rôles, et les garder séparés facilite la montée en charge par la suite :

1. Les équipements publient des mesures JSON sur des topics MQTT tels que connectmq/demo/a7f3k2/room1/telemetry.
2. Le broker achemine ces messages vers chaque abonné. Il ne conserve pas d'historique ; c'est le rôle de la base de données. Si cette distinction est nouvelle pour vous, voici en détail ce que fait un broker MQTT.
3. Node-RED s'abonne, valide et convertit chaque message en point InfluxDB, puis l'écrit. Une seconde branche interroge InfluxDB pour les graphiques.

Node-RED est un bon choix lorsque vous voulez aussi du routage, des alertes ou un tableau de bord au même endroit. Si vous avez seulement besoin d'ingérer de gros volumes, le consommateur MQTT de Telegraf est une alternative, présentée dans le guide pour stocker des données MQTT dans InfluxDB.

Étape 1 : lancer InfluxDB 2 avec Docker

L'image officielle peut initialiser une organisation, un bucket et un jeton administrateur au premier démarrage, ce qui économise plusieurs clics :

docker run -d --name influxdb -p 8086:8086 \
  -v influxdb-data:/var/lib/influxdb2 \
  -e DOCKER_INFLUXDB_INIT_MODE=setup \
  -e DOCKER_INFLUXDB_INIT_USERNAME=admin \
  -e DOCKER_INFLUXDB_INIT_PASSWORD=change-me-please \
  -e DOCKER_INFLUXDB_INIT_ORG=demo-org \
  -e DOCKER_INFLUXDB_INIT_BUCKET=sensors \
  -e DOCKER_INFLUXDB_INIT_RETENTION=30d \
  -e DOCKER_INFLUXDB_INIT_ADMIN_TOKEN=my-super-secret-token \
  influxdb:2.7

Ouvrez http://localhost:8086, connectez-vous et vous trouverez le bucket sensors avec 30 jours de rétention. En production, créez un jeton d'API distinct disposant uniquement d'un accès en écriture à ce bucket, plutôt que de confier le jeton administrateur à Node-RED. Ce tutoriel cible InfluxDB 2.x ; InfluxDB 3 modifie la partie requêtes, vérifiez donc ses notes de compatibilité avant de copier les exemples Flux.

Étape 2 : publier des données de test sur MQTT

Ce script simule deux pièces et publie sur broker.connectmq.com, le serveur MQTT public gratuit. Remplacez a7f3k2 par votre propre chaîne aléatoire : les topics du broker public sont visibles par tous, et ce serveur est réservé aux tests, pas aux données de production.

# pip install "paho-mqtt>=2.0"
import json
import random
import time
import uuid

import paho.mqtt.client as mqtt

PREFIX = "connectmq/demo/a7f3k2"
ROOMS = {"room1": 22.0, "room2": 25.0}

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                     client_id=f"influx-sim-{uuid.uuid4().hex[:8]}")
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_start()

try:
    while True:
        for room in ROOMS:
            ROOMS[room] += random.uniform(-0.2, 0.2)
            payload = {
                "temperature": round(ROOMS[room], 2),
                "humidity": round(random.uniform(38, 60), 1),
                "ts": int(time.time() * 1000),
            }
            client.publish(f"{PREFIX}/{room}/telemetry", json.dumps(payload), qos=1)
        time.sleep(5)
except KeyboardInterrupt:
    pass
finally:
    client.loop_stop()
    client.disconnect()

Inclure un horodatage côté équipement (ts) est une bonne pratique : si des messages sont retardés, la base enregistre quand même le moment de la mesure, et non celui de sa réception.

Étape 3 : installer node-red-contrib-influxdb

Dans Node-RED, ouvrez Manage palette → Install et installez node-red-contrib-influxdb. Vous obtenez trois nœuds : influxdb out pour les écritures unitaires, influxdb batch pour écrire de nombreux points d'un coup et influxdb in pour les requêtes. Tous partagent un nœud de configuration influxdb.

Configurez le nœud de configuration :

Étape 4 : associer les payloads MQTT à des fields et des tags

InfluxDB distingue les fields (les valeurs mesurées, non indexées) des tags (des métadonnées indexées sur lesquelles vous filtrez et regroupez). Bien faire ce choix est la décision de conception la plus importante de toute la configuration Node-RED InfluxDB :

• Placez les mesures numériques comme la température et l'humidité dans des fields.
• Placez les identifiants à faible cardinalité comme la pièce, la ligne ou l'identifiant d'équipement dans des tags.
• Ne mettez jamais de valeurs non bornées (horodatages, identifiants aléatoires, texte libre) dans des tags ; chaque combinaison unique de tags crée une nouvelle série, et la consommation mémoire augmente d'autant.

Ajoutez un nœud mqtt in abonné à connectmq/demo/a7f3k2/+/telemetry, avec la sortie « a parsed JSON object », puis ce nœud function. Le nœud influxdb out accepte un tableau de deux objets : les fields d'abord, les tags ensuite.

// Nœud function "to influx point"
const p = msg.payload;
if (typeof p !== "object" || p === null) {
    return null;
}

const fields = {};
if (typeof p.temperature === "number") fields.temperature = p.temperature;
if (typeof p.humidity === "number") fields.humidity = p.humidity;
if (Object.keys(fields).length === 0) {
    node.warn("No numeric fields in " + msg.topic);
    return null;
}

// connectmq/demo/a7f3k2/room1/telemetry -> room1
const room = msg.topic.split("/")[3];

// Utiliser l'horodatage de l'équipement s'il est présent (millisecondes)
if (typeof p.ts === "number") {
    fields.time = new Date(p.ts);
}

msg.payload = [fields, { room: room, source: "node-red" }];
msg.measurement = "environment";
return msg;

Reliez-le à un nœud influxdb out avec l'organisation demo-org, le bucket sensors et une précision temporelle en millisecondes. Laissez le champ measurement vide dans le nœud pour que msg.measurement soit utilisé. Déployez, attendez une minute et ouvrez le Data Explorer de l'interface InfluxDB : la measurement environment apparaît avec les fields temperature et humidity et un tag room.

Si votre version du nœud n'accepte pas de champ time dans l'objet fields, supprimez ces deux lignes ; InfluxDB utilisera alors l'heure d'écriture.

Étape 5 : interroger en Flux et tracer le résultat

L'interrogation est tout aussi simple. Un nœud inject se déclenche chaque minute vers un nœud function qui construit la requête Flux :

// Nœud function "build flux query"
msg.query = `
from(bucket: "sensors")
  |> range(start: -6h)
  |> filter(fn: (r) => r._measurement == "environment" and r._field == "temperature")
  |> aggregateWindow(every: 1m, fn: mean, createEmpty: false)
  |> keep(columns: ["_time", "_value", "room"])
`;
return msg;

Envoyez-la à un nœud influxdb in avec l'organisation demo-org. Le nœud renvoie un tableau de lignes, chacune avec _time, _value et les colonnes de tags. Convertissez-les en points de graphique, une série par pièce :

// Nœud function "rows to chart"
const rows = Array.isArray(msg.payload) ? msg.payload : [];
const byRoom = {};

for (const r of rows) {
    if (!byRoom[r.room]) byRoom[r.room] = [];
    byRoom[r.room].push({ x: Date.parse(r._time), y: Math.round(r._value * 100) / 100 });
}

const out = [{ payload: [] }];   // vider le graphique avant de le redessiner
for (const room of Object.keys(byRoom)) {
    out.push({ topic: room, payload: byRoom[room] });
}
return [out];

Reliez le tout à un ui-chart Dashboard 2.0 avec les séries issues de msg.topic, la clé X x et la clé Y y. Comme les données proviennent désormais de la base, le graphique survit automatiquement aux redémarrages. Le guide de l'historique des graphiques Node-RED compare cette approche avec des options plus légères si InfluxDB dépasse vos besoins, et le tutoriel Dashboard 2.0 explique les réglages du graphique.

Check-list de production

1. Utilisez un jeton InfluxDB dédié, avec un accès en écriture à un seul bucket pour le flow d'écriture et un accès en lecture pour le flow de requêtes.
2. Définissez une politique de rétention du bucket adaptée à la durée pendant laquelle vous avez besoin des données brutes ; sous-échantillonnez les données anciennes dans un second bucket avec une tâche InfluxDB si vous avez besoin de tendances longues.
3. Gardez des tags à faible cardinalité et cohérents d'un équipement à l'autre.
4. Validez les payloads avant écriture, comme le fait le nœud function, pour qu'un seul équipement défaillant ne puisse pas remplir le bucket de chaînes de caractères.
5. Utilisez la QoS 1 sur l'abonnement et une session persistante, afin que de courts redémarrages de Node-RED ne fassent pas perdre de mesures.
6. Utilisez le nœud influxdb batch si vous écrivez des centaines de points par seconde, en regroupant d'abord les messages avec un nœud join.
7. Surveillez les erreurs d'écriture : ajoutez un nœud catch limité au nœud influxdb out et envoyez les erreurs vers un nœud debug ou une alerte.

Dépannage

Le conflit de type de field mérite qu'on s'y arrête : InfluxDB fixe le type d'un field à la première écriture au sein d'un shard. Si un équipement envoie 22 puis 22.5, l'incohérence entier/flottant est rejetée. Comme JSON ne fait pas la différence, les capteurs qui envoient des nombres entiers déclenchent parfois ce problème. Envoyer les valeurs de façon cohérente depuis le firmware (par exemple toujours avec une décimale) et vérifier le type du field dans le Data Explorer après les premières écritures évite les mauvaises surprises.

Sécurisez aussi le côté MQTT

Une base de données protégée par des jetons soigneusement choisis ne sert à rien si n'importe qui peut publier de fausses mesures sur les topics qu'elle ingère. C'est exactement la situation sur le broker public, et c'est pourquoi il est réservé aux tests. En production, basculez le nœud de configuration mqtt-broker de Node-RED vers un broker privé avec TLS sur le port 8883 et des identifiants. Sur ConnectMQ, un broker MQTT managé, chaque broker reçoit un nom d'utilisateur, un mot de passe et un client ID générés, et les client IDs sont liés à ces identifiants : un client quelconque ne peut donc pas se faire passer pour votre flow d'ingestion.

Un broker MQTT cloud simplifie aussi le réseau : les équipements sur le terrain et Node-RED à côté d'InfluxDB se connectent tous en sortant vers le broker, vous n'avez donc pas à ouvrir de ports vers le réseau de la base de données. ConnectMQ n'écrit pas lui-même dans des bases de données ; le modèle honnête est exactement celui de cet article, un abonné Node-RED (ou Telegraf) qui écrit dans InfluxDB. Pour les bases sur les brokers, les wildcards et les nœuds de configuration, consultez le tutoriel Node-RED MQTT.

Commencez à stocker vos données MQTT avec ConnectMQ

Créez un compte gratuit sur app.connectmq.com avec l'offre Serverless, sans carte bancaire. Créez un broker, copiez les identifiants dans votre nœud de configuration Node-RED, et votre pipeline InfluxDB est alimenté par un flux privé chiffré en TLS que vous pouvez inspecter en direct dans l'explorateur de topics du portail.

Questions fréquentes

Quelles versions d'InfluxDB node-red-contrib-influxdb prend-il en charge ?

Il prend en charge InfluxDB 1.x avec InfluxQL, 1.8 avec Flux, et 2.x avec jetons, organisations et buckets. Choisissez la version dans le nœud de configuration pour que les nœuds affichent les bonnes options.

Faut-il écrire chaque message MQTT dans InfluxDB ?

Écrivez chaque mesure dont vous pourriez avoir besoin plus tard, mais sous-échantillonnez les données haute fréquence dans un bucket séparé pour les tendances de long terme. Si les équipements publient plus vite que nécessaire, calculez une moyenne dans Node-RED avant d'écrire.

Quelle est la différence entre fields et tags dans InfluxDB ?

Les fields stockent les valeurs mesurées et ne sont pas indexés. Les tags stockent des métadonnées comme la pièce ou l'identifiant d'équipement, sont indexés et doivent avoir un nombre limité de valeurs distinctes.

Puis-je utiliser Grafana au lieu d'un graphique Node-RED ?

Oui. Pointez Grafana vers le même bucket InfluxDB et utilisez la requête Flux de ce guide. Node-RED continue d'assurer l'ingestion MQTT pendant que Grafana se charge de la visualisation.