Node-RED-InfluxDB-Tutorial: MQTT-Daten speichern und abfragen

Node-RED und InfluxDB bilden zusammen eine kleine, aber vollständige Telemetrie-Pipeline: MQTT-Nachrichten kommen herein, Node-RED bringt sie in Form, und InfluxDB speichert sie als Zeitreihen, die Sie abfragen und visualisieren können. In diesem Node-RED-InfluxDB-Tutorial betreiben Sie InfluxDB 2 in Docker, schreiben MQTT-Messwerte mit sauberen Fields und Tags und fragen sie mit Flux wieder ab, um ein Chart zu zeichnen. Jeder Schritt nutzt einen kostenlosen Broker, sodass Sie die gesamte Kette in weniger als einer halben Stunde testen können. In der Fertigung ist das ein typischer Baustein für Maschinendaten, Energiemonitoring oder OEE-Kennzahlen.

Architektur: MQTT rein, Zeitreihen raus

Die Pipeline hat drei Rollen, und wer sie trennt, kann später leicht skalieren:

1. Geräte publizieren JSON-Messwerte auf MQTT-Topics wie connectmq/demo/a7f3k2/room1/telemetry.
2. Der Broker leitet diese Nachrichten an alle Abonnenten weiter. Er speichert keine Historie; das ist Aufgabe der Datenbank. Falls diese Unterscheidung neu ist: Hier steht ausführlich, was ein MQTT-Broker macht.
3. Node-RED abonniert, validiert und wandelt jede Nachricht in einen InfluxDB-Punkt um und schreibt ihn. Ein zweiter Zweig fragt InfluxDB für Charts ab.

Node-RED passt gut, wenn Sie an derselben Stelle auch Routing, Alarme oder ein Dashboard möchten. Brauchen Sie nur Ingestion mit hohem Volumen, ist der MQTT-Consumer von Telegraf eine Alternative, beschrieben im Leitfaden zum Speichern von MQTT-Daten in InfluxDB. In größeren Industrie-4.0-Architekturen abonniert der Writer typischerweise Teile eines Unified Namespace (UNS), dessen Topic-Struktur der ISA-95-Hierarchie folgt (Unternehmen, Werk, Bereich, Linie, Zelle); die Topic-Ebenen werden dann zu Tags in InfluxDB.

Schritt 1: InfluxDB 2 mit Docker starten

Das offizielle Image kann beim ersten Start eine Organisation, einen Bucket und einen Admin-Token anlegen, was einige Klicks spart:

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

Öffnen Sie http://localhost:8086, melden Sie sich an, und Sie finden den Bucket sensors mit 30 Tagen Aufbewahrung. Für die Produktion legen Sie einen separaten API-Token mit Schreibrecht nur für diesen Bucket an, statt Node-RED den Admin-Token zu geben. Dieses Tutorial zielt auf InfluxDB 2.x; InfluxDB 3 ändert die Abfrageseite, prüfen Sie also dessen Kompatibilitätshinweise, bevor Sie die Flux-Beispiele übernehmen.

Schritt 2: Testdaten an MQTT publizieren

Dieses Skript simuliert zwei Räume und publiziert an broker.connectmq.com, den kostenlosen öffentlichen MQTT-Broker. Ändern Sie a7f3k2 in eine eigene Zufallszeichenfolge; Topics auf dem öffentlichen Broker sind für alle sichtbar.

# 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()

Einen Gerätezeitstempel (ts) mitzusenden ist gute Praxis: Werden Nachrichten verzögert, speichert die Datenbank trotzdem, wann der Wert gemessen wurde, und nicht, wann er ankam.

Schritt 3: node-red-contrib-influxdb installieren

Öffnen Sie in Node-RED Palette verwalten → Installieren und installieren Sie node-red-contrib-influxdb. Sie erhalten drei Nodes: influxdb out für einzelne Schreibvorgänge, influxdb batch für viele Punkte auf einmal und influxdb in für Abfragen. Alle teilen sich einen influxdb-Konfigurations-Node.

Konfigurieren Sie den Konfigurations-Node:

Schritt 4: MQTT-Payloads auf Fields und Tags abbilden

InfluxDB unterscheidet Fields (die gemessenen Werte, nicht indiziert) von Tags (indizierte Metadaten, nach denen Sie filtern und gruppieren). Das richtig zu machen, ist die wichtigste Designentscheidung im gesamten Node-RED-InfluxDB-Setup:

• Numerische Messwerte wie Temperatur und Luftfeuchte gehören in Fields.
• Bezeichner mit geringer Kardinalität wie Raum, Linie oder Geräte-ID gehören in Tags.
• Legen Sie niemals unbegrenzte Werte (Zeitstempel, zufällige IDs, Freitext) in Tags ab; jede eindeutige Tag-Kombination erzeugt eine neue Serie, und der Speicherbedarf wächst mit.

Fügen Sie einen mqtt in-Node hinzu, der connectmq/demo/a7f3k2/+/telemetry abonniert, Ausgabe "a parsed JSON object", und danach diesen Function-Node. Der influxdb out-Node akzeptiert ein Array aus zwei Objekten: zuerst Fields, dann Tags.

// Function node "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];

// Gerätezeitstempel verwenden, falls vorhanden (Millisekunden)
if (typeof p.ts === "number") {
    fields.time = new Date(p.ts);
}

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

Verbinden Sie ihn mit einem influxdb out-Node mit Organisation demo-org, Bucket sensors und der Zeitgenauigkeit Millisekunden. Lassen Sie das Measurement im Node leer, damit msg.measurement verwendet wird. Deployen Sie, warten Sie eine Minute und öffnen Sie den Data Explorer in der InfluxDB-Oberfläche: Das Measurement environment erscheint mit den Fields temperature und humidity und einem Tag room.

Akzeptiert Ihre Node-Version kein Field time im Fields-Objekt, entfernen Sie diese beiden Zeilen; InfluxDB verwendet dann den Schreibzeitpunkt.

Schritt 5: Mit Flux abfragen und das Ergebnis visualisieren

Abfragen ist genauso einfach. Ein Inject-Node löst jede Minute einen Function-Node aus, der die Flux-Abfrage baut:

// Function node "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;

Senden Sie sie an einen influxdb in-Node mit Organisation demo-org. Der Node liefert ein Array von Zeilen, jeweils mit _time, _value und den Tag-Spalten. Wandeln Sie sie in Chart-Punkte um, eine Serie pro Raum:

// Function node "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: [] }];   // Chart vor dem Neuzeichnen leeren
for (const room of Object.keys(byRoom)) {
    out.push({ topic: room, payload: byRoom[room] });
}
return [out];

Verbinden Sie dies mit einem Dashboard-2.0-ui-chart mit Serien aus msg.topic, X-Key x und Y-Key y. Da die Daten nun aus der Datenbank kommen, übersteht das Chart Neustarts automatisch. Der Leitfaden zur Chart-Historie in Node-RED vergleicht das mit leichtgewichtigeren Optionen, falls InfluxDB mehr ist, als Sie brauchen, und das Dashboard-2.0-Tutorial erklärt die Chart-Einstellungen.

Checkliste für die Produktion

1. Verwenden Sie einen eigenen InfluxDB-Token mit Schreibrecht auf einen Bucket für den Writer-Flow und Leserecht für den Abfrage-Flow.
2. Legen Sie eine Aufbewahrungsrichtlinie für den Bucket fest, die dazu passt, wie lange Sie Rohdaten brauchen; verdichten Sie ältere Daten per InfluxDB-Task in einen zweiten Bucket, wenn Sie langfristige Trends benötigen.
3. Halten Sie Tags niedrig-kardinal und über alle Geräte konsistent.
4. Validieren Sie Payloads vor dem Schreiben, wie es der Function-Node tut, damit ein fehlerhaftes Gerät den Bucket nicht mit Strings füllen kann.
5. Verwenden Sie QoS 1 für das Abonnement und eine persistente Session, damit kurze Node-RED-Neustarts keine Messwerte kosten.
6. Nutzen Sie den influxdb batch-Node, wenn Sie Hunderte Punkte pro Sekunde schreiben, und gruppieren Sie die Nachrichten vorher mit einem Join-Node.
7. Überwachen Sie Schreibfehler: Fügen Sie einen Catch-Node mit Geltungsbereich auf den influxdb out-Node hinzu und leiten Sie Fehler an einen Debug-Node oder einen Alarm.

Fehlerbehebung

Der Field-Typkonflikt verdient Aufmerksamkeit: InfluxDB legt den Typ eines Fields beim ersten Schreiben innerhalb eines Shards fest. Sendet ein Gerät 22 und später 22.5, wird die Abweichung zwischen Integer und Float abgelehnt. Da JSON beides nicht unterscheidet, lösen Sensoren, die gelegentlich ganze Zahlen senden, das Problem aus. Werte konsequent aus der Firmware zu senden (zum Beispiel immer mit einer Nachkommastelle) und den Field-Typ nach den ersten Schreibvorgängen im Data Explorer zu prüfen, erspart spätere Überraschungen.

Auch die MQTT-Seite absichern

Eine Datenbank mit sorgfältig vergebenen Tokens nützt wenig, wenn jeder gefälschte Messwerte in die Topics publizieren kann, die sie einliest. Auf dem öffentlichen Broker ist genau das der Fall, weshalb er nur für Tests gedacht ist. Für die Produktion stellen Sie den Node-RED-Konfigurations-Node mqtt-broker auf einen privaten Broker mit TLS auf Port 8883 und Zugangsdaten um. Bei ConnectMQ erhält jeder Broker einen generierten Benutzernamen, ein Passwort und eine Client-ID, und Client-IDs sind an diese Zugangsdaten gebunden, sodass sich ein beliebiger Client nicht als Ihr Ingestion-Flow ausgeben kann.

Ein Cloud-MQTT-Broker vereinfacht außerdem das Netzwerk: Geräte im Feld und Node-RED neben InfluxDB verbinden sich beide ausgehend mit dem Broker, sodass Sie keine Ports ins Datenbanknetz öffnen müssen. ConnectMQ selbst schreibt nicht in Datenbanken; das ehrliche Muster ist genau das aus diesem Artikel, ein Node-RED- (oder Telegraf-)Abonnent, der in InfluxDB schreibt. Grundlagen zu Brokern, Wildcards und Konfigurations-Nodes finden Sie im Node-RED-MQTT-Tutorial.

MQTT-Daten mit ConnectMQ speichern

Erstellen Sie unter app.connectmq.com ein kostenloses Konto im Serverless-Plan, ohne Kreditkarte. Legen Sie einen Broker an, kopieren Sie die Zugangsdaten in Ihren Node-RED-Konfigurations-Node, und Ihre InfluxDB-Pipeline wird aus einem privaten, TLS-verschlüsselten Datenstrom gespeist, den Sie im Topic-Explorer des Portals live einsehen können.

Häufig gestellte Fragen

Welche InfluxDB-Versionen unterstützt node-red-contrib-influxdb?

Es unterstützt InfluxDB 1.x mit InfluxQL, 1.8 mit Flux und 2.x mit Tokens, Organisationen und Buckets. Wählen Sie die Version im Konfigurations-Node, damit die Nodes die passenden Optionen anzeigen.

Sollte ich jede MQTT-Nachricht in InfluxDB schreiben?

Schreiben Sie jeden Messwert, den Sie später brauchen könnten, verdichten Sie hochfrequente Daten für Langzeittrends aber in einem separaten Bucket. Publizieren Geräte schneller als nötig, mitteln Sie in Node-RED vor dem Schreiben.

Was ist der Unterschied zwischen Fields und Tags in InfluxDB?

Fields speichern die gemessenen Werte und sind nicht indiziert. Tags speichern Metadaten wie Raum oder Geräte-ID, sind indiziert und sollten nur eine begrenzte Zahl unterschiedlicher Werte haben.

Kann ich Grafana statt eines Node-RED-Charts verwenden?

Ja. Richten Sie Grafana auf denselben InfluxDB-Bucket und verwenden Sie die Flux-Abfrage aus diesem Leitfaden. Node-RED übernimmt weiter die MQTT-Ingestion, während Grafana die Visualisierung erledigt.