MQTT-Data-Historian: Industrielle Zeitreihen erfassen

Ein MQTT-Historian ist ein Subscriber, der die MQTT-Topics Ihres Werks mithört und jeden Wert als Zeitreihe speichert, damit Bediener Trends verfolgen, vergleichen und untersuchen können, was der Unified Namespace (UNS) nur im Moment zeigt. In diesem Leitfaden vergleichen Sie die wichtigsten Historian-Ansätze, richten zwei funktionierende Pipelines ein – Telegraf nach InfluxDB und Python nach TimescaleDB – und lernen die Designentscheidungen kennen, die darüber bestimmen, ob Ihr Historian auch in einem Jahr noch vertrauenswürdig ist.

Was ein MQTT-Historian macht

Ein klassischer Prozess-Historian fragt Tags aus SCADA oder SPS ab und komprimiert sie in ein proprietäres Archiv. Ein MQTT-Historian kehrt den Fluss um: Geräte und Gateways publizieren an einen Broker, und der Historian ist nur ein weiterer Subscriber. Das hat nützliche Folgen:

• Keine zusätzliche Last im Werk. Der Historian fragt nie eine SPS ab; er liest, was ohnehin publiziert wird.
• Ein einziger Integrationspunkt. Wird eine Maschine in den Namespace aufgenommen, landet sie automatisch im Historian, sofern das Abonnement ihren Zweig abdeckt.
• Kontext aus Topics. Mit ISA-95-Topics werden Standort, Bereich, Linie und Maschine ohne manuelles Mapping zu Tags oder Spalten. Im UNS bildet die ISA-95-Equipment-Hierarchie (Enterprise, Site, Area, Line, Cell) die Topic-Struktur auf dem Broker, etwa acme/werk-stuttgart/montage/linie-1/zelle-3/temperatur, und genau diese Ebenen übernimmt der Historian als Kontext.

Der Historian muss alles übernehmen, was der Broker nicht leistet: Langzeitspeicherung, effiziente Abfragen über Zeiträume, Downsampling und Aufbewahrung. Ein Broker hält den letzten Retained-Wert und kurzlebige Session-Warteschlangen, nicht Monate an Historie. Der UNS liefert also den aktuellen Zustand des Werks, der Historian dessen Vergangenheit. Diese Arbeitsteilung wird in Unified Namespace vs. Data Lake näher erläutert; wie Sie den Topic-Baum selbst strukturieren, zeigt Sparkplug B und Unified Namespace.

Historian-Ansätze im Vergleich

Es gibt nicht die eine richtige Antwort. Viele Werke betreiben einen Hot Historian mit Wochen oder Monaten hochaufgelöster Daten für Bediener und Ingenieure und schieben alles für Langzeitanalysen in einen Lake.

Beispieldaten zum Testen

Beide Pipelines erwarten Topics der Form <prefix>/<site>/<area>/<line>/<machine>/telemetry mit JSON-Payloads. Wenn Sie den Gateway-Simulator aus Unified Namespace aufbauen haben, starten Sie ihn. Andernfalls publiziert diese Schleife simulierte Daten an den öffentlichen MQTT-Broker. Er ist für Tests und Prototypen gedacht und für alle lesbar, verwenden Sie also ein eigenes zufälliges Präfix und nur simulierte Werte:

PREFIX="connectmq/demo/h6n3"
while true; do
  TS=$(date -u +%Y-%m-%dT%H:%M:%SZ)
  TEMP=$(awk -v r=$RANDOM 'BEGIN{printf "%.1f", 38 + r % 60 / 10}')
  mosquitto_pub -h broker.connectmq.com -p 1883 -i hist-sim-h6n3 \
    -t "$PREFIX/dallas/packaging/line2/filler/telemetry" \
    -m "{\"ts\":\"$TS\",\"temp_c\":$TEMP,\"speed_bpm\":118}"
  sleep 5
done

Pipeline 1: Telegraf und InfluxDB

Das MQTT-Consumer-Plugin von Telegraf abonniert den Broker und schreibt ohne eigenen Code in InfluxDB. Starten Sie InfluxDB 2 in Docker mit einem Entwicklungs-Token:

docker run -d --name influxdb -p 8086:8086 \
  -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=plant \
  -e DOCKER_INFLUXDB_INIT_BUCKET=historian \
  -e DOCKER_INFLUXDB_INIT_ADMIN_TOKEN=dev-token-change-me \
  influxdb:2.7

Erstellen Sie dann telegraf.conf. Das Topic-Parsing macht aus jeder ISA-95-Ebene einen InfluxDB-Tag:

[agent]
  interval = "10s"
  flush_interval = "10s"

[[inputs.mqtt_consumer]]
  servers = ["tcp://broker.connectmq.com:1883"]
  topics = ["connectmq/demo/h6n3/+/+/+/+/telemetry"]
  qos = 1
  client_id = "telegraf-historian-h6n3"
  name_override = "telemetry"
  data_format = "json"
  json_time_key = "ts"
  json_time_format = "2006-01-02T15:04:05Z07:00"

  [[inputs.mqtt_consumer.topic_parsing]]
    topic = "connectmq/demo/h6n3/+/+/+/+/telemetry"
    tags = "_/_/_/site/area/line/machine/_"

[[outputs.influxdb_v2]]
  urls = ["http://localhost:8086"]
  token = "dev-token-change-me"
  organization = "plant"
  bucket = "historian"

Starten Sie es mit telegraf --config telegraf.conf. Öffnen Sie die InfluxDB-Oberfläche auf Port 8086 und durchsuchen Sie den Bucket historian: Sie sehen ein Measurement telemetry mit den Tags site, area, line und machine sowie numerischen Feldern wie temp_c. Der JSON-Parser übernimmt standardmäßig numerische Felder; String-Felder listen Sie bei Bedarf explizit mit json_string_fields auf. Die Dokumentation des Telegraf MQTT Consumer führt alle Optionen auf.

Für einen privaten ConnectMQ-Broker ändern Sie den Server auf ssl://YOUR_BROKER_HOST:8883, ergänzen username = "YOUR_USERNAME" und password = "YOUR_PASSWORD" und setzen client_id auf YOUR_CLIENT_ID, da Client IDs an die im Portal ausgestellten Zugangsdaten gebunden sind. Entfernen Sie das Demo-Präfix und dessen Parsing-Ebenen, sobald Ihre Topics mit Ihrem Unternehmensnamen beginnen.

Pipeline 2: Python und TimescaleDB

Wenn Sie SQL und Joins mit Fertigungsaufträgen oder Chargendatensätzen möchten, ist TimescaleDB eine naheliegende Wahl. Starten Sie es in Docker:

docker run -d --name timescaledb -p 5432:5432 -e POSTGRES_PASSWORD=historian timescale/timescaledb:latest-pg16

Der folgende Subscriber speichert Daten im Long-Format, eine Zeile pro Metrik, was gut mit Maschinen funktioniert, die unterschiedliche Felder publizieren. Installieren Sie die Abhängigkeiten mit pip install "paho-mqtt>=2.0" "psycopg[binary]".

import json
import os
from datetime import datetime, timezone

import paho.mqtt.client as mqtt
import psycopg

DSN = os.getenv("PG_DSN", "postgresql://postgres:historian@localhost:5432/postgres")
PREFIX = os.getenv("UNS_PREFIX", "connectmq/demo/h6n3")

conn = psycopg.connect(DSN, autocommit=True)
conn.execute("""
    CREATE TABLE IF NOT EXISTS telemetry (
        time    TIMESTAMPTZ NOT NULL,
        site    TEXT, area TEXT, line TEXT, machine TEXT,
        metric  TEXT NOT NULL,
        value   DOUBLE PRECISION
    )""")
conn.execute("SELECT create_hypertable('telemetry', 'time', if_not_exists => TRUE)")

def on_connect(client, userdata, flags, reason_code, properties):
    print("historian connected:", reason_code)
    client.subscribe(f"{PREFIX}/+/+/+/+/telemetry", qos=1)

def on_message(client, userdata, msg):
    if msg.retain:
        return  # Retained-Werte wurden bereits beim ersten Publizieren gespeichert
    site, area, line, machine, _ = msg.topic[len(PREFIX) + 1:].split("/")
    try:
        data = json.loads(msg.payload)
    except ValueError:
        return
    ts = data.pop("ts", None)
    when = datetime.fromisoformat(ts.replace("Z", "+00:00")) if ts else datetime.now(timezone.utc)
    rows = [(when, site, area, line, machine, k, float(v))
            for k, v in data.items() if isinstance(v, (int, float)) and not isinstance(v, bool)]
    with conn.cursor() as cur:
        cur.executemany(
            "INSERT INTO telemetry (time, site, area, line, machine, metric, value) "
            "VALUES (%s, %s, %s, %s, %s, %s, %s)", rows)

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="pg-historian-h6n3")
client.on_connect = on_connect
client.on_message = on_message
client.connect(os.getenv("MQTT_HOST", "broker.connectmq.com"), 1883, keepalive=60)
client.loop_forever()

Fragen Sie Fünf-Minuten-Mittelwerte pro Maschine mit Standard-SQL plus time_bucket von Timescale ab:

SELECT time_bucket('5 minutes', time) AS bucket, line, machine, avg(value) AS avg_temp
FROM telemetry
WHERE metric = 'temp_c' AND time > now() - interval '1 day'
GROUP BY bucket, line, machine
ORDER BY bucket;

Entscheiden, was gespeichert wird

Jeden publizierten Wert für immer zu speichern, klingt sicher, macht Abfragen aber langsamer, und die Kosten wachsen unbemerkt. Vereinbaren Sie mit Produktion und Engineering eine Speicherrichtlinie, bevor der Historian live geht:

• Prozesswerte wie Temperaturen, Drücke und Geschwindigkeiten: einige Wochen lang roh in der publizierten Rate speichern, danach Aggregate wie Minimum, Maximum und Mittelwert pro Minute behalten.
• Energiewerte wie Wirk- und Blindleistung: in einer Auflösung speichern, die Ihr Energiemonitoring braucht, etwa für ein Energiemanagementsystem nach ISO 50001.
• Zähler wie Gut- und Ausschusszahlen: jede Änderung speichern, denn für OEE und Durchsatz zählen die Differenzen zwischen den Messwerten.
• Zustände und Alarme: jeden Übergang mit Zeitstempel speichern. Sie haben ein geringes Volumen und sind für Stillstandsanalysen unverzichtbar.
• Ereignisse wie Chargenabschlüsse: alle speichern, idealerweise in einer Tabelle mit eigenen Feldern statt als generische Metriken.
• Diagnosedaten wie Gateway-Laufzeit oder Signalstärke: kürzer aufbewahren, sofern Sie nicht gerade ein konkretes Problem untersuchen.

Wenn Sie das pro Zweig des UNS in Ihrem Topic-Standard festhalten, folgt die Historian-Konfiguration dem Namespace-Design, statt planlos zu wachsen.

Designentscheidungen, die zählen

1. Quellzeitstempel statt Ankunftszeit. Speichern Sie den Zeitstempel aus der Payload, damit gepufferte oder verzögerte Daten am richtigen Zeitpunkt landen. Behalten Sie zusätzlich die Ankunftszeit, wenn Sie Latenzen messen wollen.
2. Persistente Sessions. Geben Sie dem Historian eine feste Client ID und eine persistente Session (clean_session=False oder ein Session Expiry unter MQTT 5), damit der Broker QoS-1-Nachrichten während eines kurzen Neustarts puffert.
3. Edge-Pufferung bei langen Ausfällen. Broker-Warteschlangen überbrücken kurze Lücken. Für stundenlange WAN-Ausfälle puffern Sie am Gateway; siehe Store-and-Forward mit MQTT.
4. Umgang mit Retained Messages. Beim Abonnieren spielt der Broker Retained-Werte erneut aus. Überspringen Sie sie oder deduplizieren Sie nach Zeit und Metrik, damit Neustarts keine doppelten Zeilen erzeugen.
5. Totbänder und Report by Exception. Nur relevante Änderungen zu publizieren, reduziert den Speicherbedarf drastisch, ohne Informationen zu verlieren.
6. Aufbewahrung und Downsampling. Halten Sie Rohdaten für einen festgelegten Zeitraum und Continuous Aggregates oder heruntergetaktete Buckets für länger vor.
7. Horizontale Skalierung. Mit Shared Subscriptions unter MQTT 5 können sich mehrere Historian-Worker $share/historian/... teilen, sodass jede Nachricht genau einmal geschrieben wird. Bei ConnectMQ sind Shared Subscriptions ab dem kostenpflichtigen Serverless-Plan verfügbar.

ConnectMQ enthält keine eingebaute Datenbank und keinen Historian-Connector und braucht für dieses Muster auch keinen: Jeder der obigen Subscriber verbindet sich als normaler MQTT-Client. Mehr zur Modellierung der Daten selbst lesen Sie unter industrielle Zeitreihendaten, die übergreifende Architektur finden Sie im Leitfaden zu Industrial IoT mit MQTT.

Speisen Sie Ihren MQTT-Historian aus ConnectMQ

Ein Historian ist nur so vollständig wie der Datenstrom, der ihn speist. ConnectMQ ist ein verwalteter MQTT-Broker mit MQTT 3.1.1 und 5.0, QoS 1 und 2, persistenten Sessions, TLS und einem ISA-95-fähigen Portal, sodass Ihr Telegraf- oder Python-Historian eine verlässliche Quelle hat. Erstellen Sie einen Broker im kostenlosen Serverless-Plan unter app.connectmq.com, ohne Kreditkarte, oder erfahren Sie mehr über den Betrieb eines Cloud-MQTT-Brokers für Ihr Werk.

Häufig gestellte Fragen

Ist ein MQTT-Broker ein Historian?

Nein. Ein Broker leitet Nachrichten weiter und hält pro Topic den letzten Retained-Wert. Ein Historian abonniert den Broker und speichert die vollständige Zeitreihe für Abfragen.

Sollte ich InfluxDB oder TimescaleDB wählen?

Wählen Sie InfluxDB mit Telegraf für das schnellste Setup rein per Konfiguration und für Grafana-Dashboards. Wählen Sie TimescaleDB, wenn Sie SQL und Joins mit relationalen Daten wie Aufträgen und Chargen möchten.

Kann ein kommerzieller Historian aus MQTT lesen?

Viele kommerzielle Historians bieten MQTT- oder Sparkplug-Connectoren. Prüfen Sie, ob der Connector Ihr Payload-Format und Ihre Topic-Struktur unterstützt und ob er Sparkplug-Births und Aliases verarbeiten kann, falls Sie diese nutzen.

Wie vermeide ich Datenverlust, wenn der Historian neu startet?

Verwenden Sie eine feste Client ID mit persistenter Session und QoS 1, damit der Broker Nachrichten kurzzeitig puffert, und puffern Sie für längere Ausfälle an der Edge.