Node-RED MySQL : enregistrer des données MQTT dans MySQL ou MariaDB avec des requêtes paramétrées

Node-RED et MySQL forment un duo pratique lorsque vous voulez stocker des données de capteurs MQTT dans une base que votre équipe sait déjà interroger, sauvegarder et exploiter dans ses rapports. Dans ce tutoriel, vous allez créer une table pour les mesures, vous abonner à des topics MQTT dans Node-RED, insérer chaque message avec des requêtes paramétrées, passer aux insertions groupées lorsque le débit augmente et mettre en place une rétention pour que la table ne grossisse pas indéfiniment. Tout fonctionne de la même manière avec MariaDB.

Pourquoi enregistrer des données MQTT dans MySQL ?

Un broker MQTT transporte des messages ; ce n'est pas un historien. Les retained messages vous donnent la dernière valeur par topic, mais pas les valeurs de la semaine dernière. Pour produire des rapports, comparer des équipes ou alimenter un outil de BI, l'historique doit être conservé dans un endroit durable.

Les bases de séries temporelles comme InfluxDB sont un choix naturel pour la télémétrie à haut débit, et nous les traitons dans Node-RED avec InfluxDB. MySQL et MariaDB restent gagnants dans bien des situations : l'entreprise les exploite déjà, les outils de reporting s'y connectent nativement, vous devez croiser les mesures avec des données relationnelles comme des ordres de fabrication ou des listes d'actifs, ou le débit de données est modeste. Quelques centaines de mesures par minute restent un terrain confortable pour une table bien indexée.

Étape 1 : créer la base et la table

Si vous n'avez pas encore de serveur de base de données, lancez MariaDB sous Docker pour vos tests :

docker run -d --name iot-mariadb -p 3306:3306 \
  -e MARIADB_ROOT_PASSWORD=change-root \
  -e MARIADB_DATABASE=iot \
  -e MARIADB_USER=nodered \
  -e MARIADB_PASSWORD=change-me \
  mariadb:11

Créez ensuite la table. Une ligne par mesure, avec le topic d'origine conservé pour la traçabilité et le payload complet conservé pour tout ce que vous n'auriez pas anticipé :

CREATE TABLE mqtt_readings (
  id          BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
  received_at DATETIME(3)  NOT NULL,
  topic       VARCHAR(255) NOT NULL,
  device      VARCHAR(64)  NOT NULL,
  metric      VARCHAR(64)  NOT NULL,
  value       DOUBLE       NULL,
  payload     TEXT         NULL,
  INDEX idx_device_metric_time (device, metric, received_at),
  INDEX idx_time (received_at)
);

Stockez les horodatages en UTC et convertissez-les dans vos rapports. Mélanger heure locale et changements d'heure dans une table de séries temporelles est une source sûre de trous déroutants et d'heures en double.

N'accordez à l'utilisateur Node-RED que ce dont il a besoin :

GRANT SELECT, INSERT, DELETE ON iot.mqtt_readings TO 'nodered'@'%';

Étape 2 : installer le nœud MySQL de Node-RED

Installez node-red-node-mysql depuis Menu → Manage palette. Il ajoute un nœud mysql avec une configuration de base de données : hôte, port (3306), utilisateur, mot de passe, nom de la base et fuseau horaire. Réglez le fuseau horaire sur Z ou +00:00 pour que les dates soient interprétées en UTC.

Le nœud suit un contrat simple :

• msg.topic contient l'instruction SQL.
• msg.payload contient les paramètres : un tableau pour les marqueurs ?, ou un objet pour les marqueurs nommés :name.
• Le résultat revient dans msg.payload : des lignes pour un SELECT, ou un objet résultat avec affectedRows et insertId pour un INSERT.

Étape 3 : s'abonner à MQTT

Ajoutez un nœud mqtt in connecté à votre broker. Pour les tests, utilisez le serveur MQTT public gratuit broker.connectmq.com, port 1883, avec un client ID unique, et abonnez-vous à connectmq/demo/db91kz/+/+ en QoS 1 avec la sortie en auto-détection. Les deux niveaux de wildcard capturent l'équipement et la métrique, par exemple connectmq/demo/db91kz/boiler1/temperature. Remplacez db91kz par votre propre chaîne aléatoire ; tout ce qui passe par le broker public est visible des autres, il est donc réservé aux tests.

Étape 4 : insérer avec des requêtes paramétrées

Voici la règle la plus importante de ce tutoriel : ne construisez jamais du SQL en concaténant des payloads MQTT dans une chaîne. Les topics et les payloads proviennent des équipements et, sur un broker partagé ou mal configuré, potentiellement de n'importe qui. Un payload tel que 1); DROP TABLE mqtt_readings; -- doit finir comme un texte inoffensif dans une colonne, pas être exécuté.

Ajoutez un nœud function entre le nœud mqtt in et le nœud mysql :

// Topic : connectmq/demo/db91kz/<device>/<metric>
const mqttTopic = msg.topic;
const parts = mqttTopic.split("/");
if (parts.length !== 5) return null;

const device = parts[3].substring(0, 64);
const metric = parts[4].substring(0, 64);

// Accepter des nombres, des chaînes numériques ou des objets {"value": ...}
let raw = msg.payload;
let value = null;
if (typeof raw === "number") {
    value = raw;
} else if (typeof raw === "string" && raw.trim() !== "" && !isNaN(Number(raw))) {
    value = Number(raw);
} else if (typeof raw === "object" && raw !== null && typeof raw.value === "number") {
    value = raw.value;
}

const payloadText = typeof raw === "string" ? raw : JSON.stringify(raw);
// Format MySQL DATETIME(3) en UTC : 2026-10-16 08:15:30.123
const receivedAt = new Date().toISOString().slice(0, 23).replace("T", " ");

msg.topic =
    "INSERT INTO mqtt_readings (received_at, topic, device, metric, value, payload) " +
    "VALUES (?, ?, ?, ?, ?, ?)";
msg.payload = [receivedAt, mqttTopic, device, metric, value, payloadText];
return msg;

Notez que msg.topic est écrasé par l'instruction SQL : c'est pourquoi le code sauvegarde d'abord le topic MQTT dans mqttTopic pour la colonne topic. Reliez la fonction au nœud mysql et ajoutez un nœud debug après celui-ci pour voir l'insertId de chaque ligne.

Tester l'insertion

mosquitto_pub -h broker.connectmq.com -p 1883 -i sim-db91kz -t "connectmq/demo/db91kz/boiler1/temperature" -m "71.4"
mosquitto_pub -h broker.connectmq.com -p 1883 -i sim-db91kz -t "connectmq/demo/db91kz/boiler1/pressure" -m '{"value": 2.3, "unit": "bar"}'
mosquitto_pub -h broker.connectmq.com -p 1883 -i sim-db91kz -t "connectmq/demo/db91kz/boiler1/state" -m "running"

Vérifiez ensuite la table :

SELECT received_at, device, metric, value, payload
FROM mqtt_readings
ORDER BY id DESC
LIMIT 10;

Le troisième message est enregistré avec value à NULL et running dans la colonne payload, ce qui est exactement ce que vous voulez pour les états et autres données non numériques.

Étape 5 : insertions groupées pour les débits élevés

Un INSERT par message convient pour quelques messages par seconde. Au-delà, chaque aller-retour et chaque commit s'additionnent. Le regroupement réduit la charge à la fois sur Node-RED et sur la base.

Modifiez la fonction de l'étape 4 pour qu'elle produise un tableau de ligne au lieu d'une instruction SQL (définissez msg.payload = [receivedAt, ...] et ne touchez pas à msg.topic), puis ajoutez un nœud join en mode manuel : regroupez chaque msg.payload dans un array, envoi après 100 messages ou après un délai de 5 secondes. Vérifiez avec un nœud debug que le résultat est un tableau de lignes (un tableau de tableaux) et non un seul tableau plat. Faites-le suivre de cette fonction :

const rows = msg.payload;            // tableau de tableaux à 6 éléments
if (!Array.isArray(rows) || rows.length === 0) return null;

const placeholders = rows.map(() => "(?, ?, ?, ?, ?, ?)").join(", ");
msg.topic =
    "INSERT INTO mqtt_readings (received_at, topic, device, metric, value, payload) VALUES " +
    placeholders;
msg.payload = rows.flat();
return msg;

Chaque valeur reste transmise comme paramètre : le lot est donc aussi sûr que l'insertion unitaire. Si Node-RED s'arrête alors qu'un lot est en attente, jusqu'à cinq secondes de données sont perdues ; pour la plupart des usages de supervision, c'est acceptable, et pour tout besoin plus strict, conservez les insertions unitaires.

Étape 6 : rétention et requêtes utiles

Décidez combien de temps les mesures brutes sont conservées. Un nettoyage quotidien simple dans Node-RED utilise un nœud inject réglé pour se répéter à heure fixe, suivi d'un nœud function :

msg.topic = "DELETE FROM mqtt_readings WHERE received_at < ? LIMIT 50000";
const cutoff = new Date(Date.now() - 90 * 24 * 3600 * 1000);
msg.payload = [cutoff.toISOString().slice(0, 23).replace("T", " ")];
return msg;

Le LIMIT garde chaque suppression petite afin de ne pas verrouiller la table longtemps ; exécutez-la plus souvent si une journée produit davantage de lignes. Le planificateur d'événements de MySQL peut faire le même travail au sein de la base si vous préférez.

Deux requêtes que vous utiliserez en permanence :

-- Moyenne horaire par équipement sur les dernières 24 heures
SELECT device,
       DATE_FORMAT(received_at, '%Y-%m-%d %H:00') AS hour,
       AVG(value) AS avg_value,
       MAX(value) AS max_value
FROM mqtt_readings
WHERE metric = 'temperature'
  AND received_at >= UTC_TIMESTAMP() - INTERVAL 24 HOUR
GROUP BY device, hour
ORDER BY device, hour;

-- Équipements qui n'ont rien envoyé depuis 15 minutes
SELECT device, MAX(received_at) AS last_seen
FROM mqtt_readings
GROUP BY device
HAVING last_seen < UTC_TIMESTAMP() - INTERVAL 15 MINUTE;

La seconde requête est un moyen peu coûteux de repérer les équipements silencieux. Pour être prévenu au lieu de devoir vérifier, combinez-la avec le flow d'alerte e-mail Node-RED. Pour afficher l'historique sur un tableau de bord, le guide de l'historique des graphiques Node-RED montre comment l'interroger et le tracer.

Dépanner les flows Node-RED MySQL

Configuration de production

Pour des données réelles, quittez le broker public. Créez un broker privé sur ConnectMQ, puis configurez le nœud broker de Node-RED avec le port 8883, TLS activé et les valeurs YOUR_USERNAME, YOUR_PASSWORD et YOUR_CLIENT_ID issues du portail. Utilisez une session persistante (clean session désactivé) avec la QoS 1, afin que de courts redémarrages de Node-RED ne fassent pas perdre les messages que le broker peut encore livrer ; sur l'offre gratuite Serverless, la QoS va jusqu'à 1, ce qui est de toute façon le bon niveau pour la journalisation.

Ce partage des responsabilités est délibéré. ConnectMQ est un broker MQTT managé qui livre des messages ; il n'écrit pas lui-même dans des bases de données. Node-RED est le client tourné vers la base, ce qui laisse votre schéma, votre rétention et votre SQL entièrement sous votre contrôle. Le tutoriel Node-RED MQTT couvre les autres réglages du nœud broker.

Enregistrez vos données MQTT avec Node-RED et ConnectMQ

Créez un broker avec l'offre gratuite Serverless sur app.connectmq.com, sans carte bancaire, connectez Node-RED via TLS et commencez à constituer un historique que vous pouvez interroger en SQL standard. L'explorateur de topics en direct du portail vous aide à confirmer les topics et les formats de payload avant de concevoir votre table autour d'eux. Si vous hésitez encore entre plusieurs options d'hébergement, notre guide du broker MQTT cloud compare les approches.

Questions fréquentes

Comment passer des paramètres au nœud MySQL de Node-RED ?

Placez le SQL avec des marqueurs ? dans msg.topic et un tableau de valeurs dans msg.payload. Le nœud accepte aussi des marqueurs nommés :name lorsque msg.payload est un objet.

Cela fonctionne-t-il avec MariaDB ?

Oui. node-red-node-mysql se connecte à MySQL comme à MariaDB, et la table, les insertions et les requêtes de ce tutoriel fonctionnent sans modification sur l'un ou l'autre.

Faut-il utiliser MySQL ou InfluxDB pour les données MQTT ?

Utilisez MySQL ou MariaDB lorsque vous avez besoin de jointures avec des données relationnelles, d'outils de reporting existants ou de débits modérés. Utilisez une base de séries temporelles pour la télémétrie à haut débit et une rétention longue avec sous-échantillonnage.

ConnectMQ peut-il stocker directement les messages dans une base de données ?

Non. ConnectMQ est un broker MQTT sans intégration de base de données. Faites tourner Node-RED ou un autre abonné qui se connecte au broker et écrit dans la base, comme le montre ce tutoriel.