Node-RED MySQL: MQTT-Daten mit parametrisierten Abfragen in MySQL oder MariaDB speichern

Node-RED und MySQL sind ein praktisches Paar, wenn MQTT-Sensordaten in einer Datenbank landen sollen, die Ihr Team bereits abfragen, sichern und auswerten kann. In diesem Tutorial legen Sie eine Tabelle für Messwerte an, abonnieren MQTT-Topics in Node-RED, fügen jede Nachricht mit parametrisierten Abfragen ein, wechseln bei steigender Nachrichtenrate zu Batch-Inserts und richten eine Aufbewahrung ein, damit die Tabelle nicht endlos wächst. Alles funktioniert mit MariaDB genauso.

Warum MQTT-Daten in MySQL speichern?

Ein MQTT-Broker transportiert Nachrichten; er ist kein Historian. Retained Messages liefern den letzten Wert pro Topic, aber nicht die Werte der vergangenen Woche. Um Berichte zu erstellen, Schichten zu vergleichen oder ein BI-Werkzeug zu versorgen, brauchen Sie die Historie an einem dauerhaften Ort.

Zeitreihendatenbanken wie InfluxDB sind die naheliegende Wahl für hochfrequente Telemetrie; das behandeln wir in Node-RED mit InfluxDB. MySQL und MariaDB gewinnen dennoch in vielen Situationen: Das Unternehmen betreibt sie bereits, Reporting-Werkzeuge verbinden sich nativ, Sie müssen Messwerte mit relationalen Daten wie Fertigungsaufträgen oder Anlagenlisten aus MES und ERP verknüpfen, oder die Datenrate ist moderat. Einige Hundert Messwerte pro Minute sind für eine gut indizierte Tabelle bequemes Terrain.

Schritt 1: Datenbank und Tabelle anlegen

Haben Sie noch keinen Datenbankserver, starten Sie MariaDB zum Testen in Docker:

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

Legen Sie dann die Tabelle an. Eine Zeile pro Messwert, mit dem ursprünglichen Topic zur Rückverfolgbarkeit und dem vollständigen Payload für alles, was Sie nicht vorhergesehen haben:

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)
);

Speichern Sie Zeitstempel in UTC und rechnen Sie in Ihren Berichten um. Lokalzeit und Sommerzeitumstellungen in einer Zeitreihentabelle zu mischen, ist eine zuverlässige Quelle für verwirrende Lücken und doppelte Stunden.

Geben Sie dem Node-RED-Benutzer nur die Rechte, die er braucht:

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

Schritt 2: Den Node-RED-MySQL-Node installieren

Installieren Sie node-red-node-mysql über Menü → Palette verwalten. Er fügt einen mysql-Node mit einer Datenbankkonfiguration hinzu: Host, Port (3306), Benutzer, Passwort, Datenbankname und Zeitzone. Stellen Sie die Zeitzone auf Z oder +00:00, damit Datumswerte als UTC interpretiert werden.

Der Node folgt einem einfachen Vertrag:

• msg.topic enthält die SQL-Anweisung.
• msg.payload enthält die Parameter: ein Array für ?-Platzhalter oder ein Objekt für benannte :name-Platzhalter.
• Das Ergebnis kommt in msg.payload zurück: Zeilen bei einem SELECT oder ein Ergebnisobjekt mit affectedRows und insertId bei einem INSERT.

Schritt 3: MQTT abonnieren

Fügen Sie einen mqtt in-Node hinzu, der mit Ihrem Broker verbunden ist. Verwenden Sie zum Testen den öffentlichen MQTT-Broker unter broker.connectmq.com, Port 1883, mit einer eindeutigen Client-ID, und abonnieren Sie connectmq/demo/db91kz/+/+ mit QoS 1 und der Ausgabe auto-detect. Die beiden Wildcard-Ebenen erfassen Gerät und Messgröße, zum Beispiel connectmq/demo/db91kz/boiler1/temperature. Ersetzen Sie db91kz durch eine eigene Zufallszeichenfolge; alles auf dem öffentlichen Broker ist für andere sichtbar.

Schritt 4: Mit parametrisierten Abfragen einfügen

Hier steht die wichtigste Regel dieses Tutorials: Bauen Sie SQL niemals, indem Sie MQTT-Payloads in einen String verketten. Topics und Payloads stammen von Geräten und auf einem geteilten oder falsch konfigurierten Broker potenziell von jedermann. Ein Payload wie 1); DROP TABLE mqtt_readings; -- soll als harmloser Text in einer Spalte landen und nicht ausgeführt werden.

Fügen Sie zwischen mqtt in-Node und mysql-Node einen Function-Node ein:

// 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);

// Zahlen, numerische Strings oder {"value": ...}-Objekte akzeptieren
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);
// MySQL-DATETIME(3)-Format in 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;

Beachten Sie, dass msg.topic mit der SQL-Anweisung überschrieben wird; deshalb speichert der Code das MQTT-Topic zuerst in mqttTopic für die Spalte topic. Verbinden Sie die Funktion mit dem mysql-Node und fügen Sie danach einen Debug-Node hinzu, um die insertId jeder Zeile zu sehen.

Den Insert testen

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"

Prüfen Sie dann die Tabelle:

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

Die dritte Nachricht landet mit value gleich NULL und running in der Payload-Spalte, genau so, wie Sie es für Zustände und andere nicht numerische Daten wollen.

Schritt 5: Batch-Inserts für höhere Nachrichtenraten

Ein INSERT pro Nachricht ist bei einigen Nachrichten pro Sekunde in Ordnung. Darüber summieren sich jeder Roundtrip und jeder Commit. Batching senkt die Last sowohl für Node-RED als auch für die Datenbank.

Ändern Sie die Funktion aus Schritt 4 so, dass sie statt einer SQL-Anweisung ein Zeilen-Array ausgibt (setzen Sie msg.payload = [receivedAt, ...] und lassen Sie msg.topic unverändert), und fügen Sie dann einen join-Node im manuellen Modus hinzu: jedes msg.payload zu einem Array zusammenfassen, nach 100 Nachrichten oder nach einem Timeout von 5 Sekunden senden. Prüfen Sie mit einem Debug-Node, dass das Ergebnis ein Array von Zeilen (ein Array aus Arrays) ist und nicht ein einziges flaches Array. Danach folgt diese Funktion:

const rows = msg.payload;            // Array aus Arrays mit je 6 Elementen
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;

Jeder Wert wird weiterhin als Parameter übergeben, der Batch ist also genauso sicher wie der einzelne Insert. Stoppt Node-RED, während ein Batch aussteht, gehen bis zu fünf Sekunden Daten verloren; für das meiste Monitoring ist das akzeptabel, für strengere Anforderungen bleiben Sie bei einzelnen Inserts.

Schritt 6: Aufbewahrung und nützliche Abfragen

Legen Sie fest, wie lange Rohwerte aufbewahrt werden. Eine einfache tägliche Bereinigung in Node-RED nutzt einen inject-Node, der zu einer festen Uhrzeit auslöst, gefolgt von einem Function-Node:

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;

Das LIMIT hält jeden Löschvorgang klein, damit er die Tabelle nicht lange sperrt; führen Sie ihn häufiger aus, wenn ein Tag mehr Zeilen erzeugt. Der Event Scheduler von MySQL kann dieselbe Aufgabe auch innerhalb der Datenbank erledigen, wenn Ihnen das lieber ist.

Zwei Abfragen, die Sie ständig brauchen werden:

-- Stündlicher Mittelwert pro Gerät für die letzten 24 Stunden
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;

-- Geräte, die sich seit 15 Minuten nicht gemeldet haben
SELECT device, MAX(received_at) AS last_seen
FROM mqtt_readings
GROUP BY device
HAVING last_seen < UTC_TIMESTAMP() - INTERVAL 15 MINUTE;

Die zweite Abfrage ist ein günstiger Weg, stumme Geräte zu finden. Um benachrichtigt zu werden, statt nachsehen zu müssen, kombinieren Sie sie mit dem Node-RED-Flow für E-Mail-Alarme. Um die Historie in einem Dashboard zu zeigen, erklärt der Leitfaden zur Chart-Historie in Node-RED, wie Sie sie abfragen und darstellen.

Fehlerbehebung bei Node-RED-MySQL-Flows

Produktionssetup

Für echte Daten verlassen Sie den öffentlichen Broker. Legen Sie bei ConnectMQ einen privaten Broker an und konfigurieren Sie den Node-RED-Broker-Node mit Port 8883, aktiviertem TLS und den Werten YOUR_USERNAME, YOUR_PASSWORD und YOUR_CLIENT_ID aus dem Portal. Verwenden Sie eine persistente Session (Clean Session aus) mit QoS 1, damit kurze Node-RED-Neustarts keine Nachrichten kosten, die der Broker noch zustellen kann; im kostenlosen Serverless-Plan geht QoS bis 1, was für das Logging ohnehin die richtige Stufe ist.

Diese Aufgabenteilung ist gewollt. ConnectMQ ist ein vollständig verwalteter MQTT-Broker, der Nachrichten zustellt; er schreibt selbst nicht in Datenbanken. Node-RED ist der Client auf der Datenbankseite, wodurch Schema, Aufbewahrung und SQL vollständig unter Ihrer Kontrolle bleiben. Das Node-RED-MQTT-Tutorial behandelt die übrigen Einstellungen des Broker-Nodes.

MQTT-Daten mit Node-RED und ConnectMQ loggen

Erstellen Sie unter app.connectmq.com einen Broker im kostenlosen Serverless-Plan, ohne Kreditkarte, verbinden Sie Node-RED per TLS und beginnen Sie mit einer Historie, die Sie mit einfachem SQL abfragen können. Der Live-Topic-Explorer im Portal hilft Ihnen, Topics und Payload-Formate zu bestätigen, bevor Sie Ihre Tabelle danach entwerfen. Wenn Sie die Hosting-Optionen noch abwägen, vergleicht unser Leitfaden zum Cloud-MQTT-Broker die Ansätze.

Häufig gestellte Fragen

Wie übergebe ich Parameter an den Node-RED-MySQL-Node?

Legen Sie das SQL mit ?-Platzhaltern in msg.topic und ein Array mit Werten in msg.payload ab. Der Node akzeptiert auch benannte :name-Platzhalter, wenn msg.payload ein Objekt ist.

Funktioniert das auch mit MariaDB?

Ja. node-red-node-mysql verbindet sich sowohl mit MySQL als auch mit MariaDB, und Tabelle, Inserts und Abfragen aus diesem Tutorial laufen auf beiden unverändert.

Sollte ich für MQTT-Daten MySQL oder InfluxDB verwenden?

Nutzen Sie MySQL oder MariaDB, wenn Sie Joins mit relationalen Daten, bestehende Reporting-Werkzeuge oder moderate Datenraten haben. Nutzen Sie eine Zeitreihendatenbank für hochfrequente Telemetrie und lange Aufbewahrung mit Downsampling.

Kann ConnectMQ Nachrichten direkt in einer Datenbank speichern?

Nein. ConnectMQ ist ein MQTT-Broker ohne Datenbankintegrationen. Betreiben Sie Node-RED oder einen anderen Abonnenten, der sich mit dem Broker verbindet und in die Datenbank schreibt, wie in diesem Tutorial gezeigt.