Industrielle Zeitreihendaten sind der Strom zeitgestempelter Messwerte, den Maschinen, Sensoren und Steuerungen in jeder Sekunde jeder Schicht erzeugen. Sie gut zu handhaben bedeutet zu entscheiden, wie oft abgetastet wird, was verworfen wird, wie Zeitstempel gesetzt werden und wo die Daten liegen. Dieser Artikel geht jede dieser Entscheidungen durch und liefert lauffähigen Code, um Totband-Kompression auf MQTT-Daten anzuwenden, bevor sie einen Historian erreichen.
Was industrielle Zeitreihendaten besonders macht
Alle Zeitreihen haben dieselbe Form: einen Zeitstempel, eine Kennung und einen Wert. Industrielle Daten bringen einige Komplikationen mit, die IT-Monitoringdaten selten haben.
- Sehr unterschiedliche Änderungsraten. Ein Tankfüllstand ändert sich langsam über Stunden; ein Schwingungssignal ändert sich tausendfach pro Sekunde. Eine einzige Abtastrichtlinie passt nicht zu beiden.
- Lange Aufbewahrungsfristen. Qualitätsuntersuchungen, regulatorische Aufzeichnungen und Modelltraining, etwa für Predictive Maintenance, benötigen oft Jahre an Historie, nicht Wochen.
- Qualität ist so wichtig wie der Wert. Ein Wert von einem getrennten Sensor ist keine Messung. Industrielle Systeme führen zu jedem Punkt ein Qualitätsflag mit.
- Ereignis- und Zustandsdaten gemischt mit Analogdaten. Maschinenzustände, Alarme, Chargen-IDs und Rezeptwechsel sind spärlich und müssen mit dichten Analogsignalen verknüpft werden.
- Uhren, die Sie nicht kontrollieren. SPS, Gateways und Server haben jeweils eigene Uhren, und diese driften.
Diese Punkte von Anfang an richtig zu machen, ist weit günstiger, als später Jahre an inkonsistenter Historie zu reparieren.
Abtastraten wählen
Die Abtastrate sollte der Physik des Signals und der Frage folgen, die Sie beantworten möchten, nicht der Standard-Zykluszeit der SPS.
| Signaltyp | Typisches Verhalten | Praktischer Ansatz |
|---|---|---|
| Temperaturen, Füllstände, langsame Drücke | Ändern sich über Sekunden bis Minuten | Alle 1–10 s abtasten, mit Totband komprimieren |
| Durchflüsse, Motorstrom, Drehzahlen | Ändern sich im Subsekunden- bis Sekundenbereich | Alle 100 ms–1 s abtasten, mit Totband komprimieren |
| Maschinenzustand, Alarme, Zähler | Diskrete Ereignisse | Nur bei Änderung publizieren (Report by Exception) |
| Schwingung, Akustik, Netzqualität | Ändern sich innerhalb von Millisekunden | An der Edge im kHz-Bereich abtasten, Merkmale publizieren (RMS, Spitze, Frequenzbänder) |
Zwei Regeln helfen. Erstens das Nyquist-Prinzip: Um eine Schwankung zu sehen, tasten Sie mindestens doppelt so schnell ab wie diese Schwankung, in der Praxis mehrfach so schnell. Zweitens: Streamen Sie hochfrequente Roh-Wellenformen nicht dauerhaft über MQTT. Berechnen Sie Merkmale an der Edge und publizieren Sie diese; Roh-Schnappschüsse senden Sie nur auf Anforderung oder bei Ereignissen.
Für diskrete Daten ist Report by Exception das natürliche Muster. Der Artikel zu Report by Exception erklärt, wie es den Datenverkehr reduziert, ohne Informationen zu verlieren.
Kompression: Das Signal behalten, das Rauschen verwerfen
Kompression bedeutet in industriellen Historians meist verlustbehaftetes Filtern bei der Erfassung, nicht Zip-Dateien. Ziel ist es, weniger Punkte zu speichern und das Signal dennoch innerhalb einer bekannten Toleranz rekonstruieren zu können.
Totband-Filterung (Exception)
Die einfachste Methode: Ein neuer Punkt wird nur gespeichert, wenn der Wert um mehr als einen Schwellwert vom zuletzt gespeicherten Wert abweicht. Ein Totband von 0,2 degC auf einer Temperatur bedeutet, dass Schwankungen unter 0,2 degC nicht gespeichert werden. Die meisten Implementierungen speichern zudem nach einer maximalen Zeitspanne einen Punkt, auch wenn sich nichts geändert hat (ein Heartbeat), damit Consumer ein stabiles Signal von einem toten unterscheiden können.
Swinging-Door-Kompression
Swinging Door Trending, das viele kommerzielle Historians verwenden, behält einen Punkt nur dann, wenn eine Gerade vom zuletzt gespeicherten Punkt die Zwischenwerte nicht mehr innerhalb der Toleranz abbilden kann. Es erhält Steigungen deutlich besser als ein einfaches Totband, sodass Rampen mit wenigen statt vielen Punkten gespeichert werden.
Toleranzen wählen
Legen Sie die Toleranz relativ zur Genauigkeit des Sensors fest, nicht als beliebigen Prozentsatz. Filtern unterhalb des Rauschpegels des Sensors speichert Rauschen; Filtern weit darüber verliert echte Prozessänderungen. Dokumentieren Sie die Toleranz pro Tag-Typ in Ihrem IIoT-Datenmodell, damit sie wie jedes andere Modellattribut geprüft wird.
Totband-Kompression auf MQTT-Daten in Python
Dieser Subscriber liest JSON-Telemetrie aus MQTT, wendet ein Totband mit Heartbeat an und schreibt die verbleibenden Punkte im InfluxDB-Line-Protocol auf die Standardausgabe, die Sie in eine Datei umleiten oder an einen Writer weiterleiten können. Er nutzt den kostenlosen öffentlichen MQTT-Broker und enthält einen eingebauten Demo-Simulator.
import json
import math
import random
import threading
import time
import uuid
import paho.mqtt.client as mqtt
BROKER = "broker.connectmq.com"
PREFIX = f"connectmq/demo/{uuid.uuid4().hex[:8]}/plant1/line1"
DEADBAND = {"temperature": 0.2, "pressure": 0.05}
HEARTBEAT_S = 60
last_stored = {} # Topic -> (Wert, ts_Sekunden)
def keep(topic, metric, value, ts):
prev = last_stored.get(topic)
if prev is None:
return True
band = DEADBAND.get(metric, 0.0)
return abs(value - prev[0]) > band or ts - prev[1] >= HEARTBEAT_S
def on_connect(client, userdata, flags, reason_code, properties):
client.subscribe(f"{PREFIX}/+/+", qos=1)
def on_message(client, userdata, msg):
asset, metric = msg.topic.split("/")[-2:]
p = json.loads(msg.payload)
ts = p["ts"] / 1000
if keep(msg.topic, metric, p["value"], ts):
last_stored[msg.topic] = (p["value"], ts)
print(f'{metric},asset={asset},quality={p["quality"]} value={p["value"]} {int(p["ts"] * 1e6)}')
def simulate(pub):
t0 = time.time()
while True:
t = time.time() - t0
temp = 60 + 5 * math.sin(t / 120) + random.gauss(0, 0.05)
pres = 2.5 + random.gauss(0, 0.01)
ts = int(time.time() * 1000)
for metric, val in [("temperature", temp), ("pressure", pres)]:
body = {"value": round(val, 3), "ts": ts, "quality": "GOOD"}
pub.publish(f"{PREFIX}/pump1/{metric}", json.dumps(body), qos=0)
time.sleep(1)
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="tsdb-" + uuid.uuid4().hex[:8])
client.on_connect = on_connect
client.on_message = on_message
client.connect(BROKER, 1883, keepalive=60)
threading.Thread(target=simulate, args=(client,), daemon=True).start()
client.loop_forever()
Lassen Sie es einige Minuten laufen und zählen Sie die Zeilen: Der Druck, der nur Rauschen unterhalb seines Totbands zeigt, erzeugt etwa einen Punkt pro Minute (den Heartbeat), während die Temperatur Punkte erzeugt, sobald sich die Sinuskurve bewegt. Genau dieses Verhalten wollen Sie von einem Historian: dicht, wo sich der Prozess bewegt, spärlich, wo er es nicht tut.
Zeitstempel und Reihenfolge
Setzen Sie Zeitstempel immer an der Quelle, möglichst nah an der Messung, in UTC mit Millisekunden. Die Ankunftszeit am Broker ist nicht die Messzeit, besonders wenn ein Edge-Gateway Daten während eines Netzwerkausfalls puffert und später weiterleitet. Das Muster Store-and-Forward ist auf Quellzeitstempel angewiesen, um verspätete Daten an der richtigen Stelle einzuordnen.
Synchronisieren Sie die Uhren auf Gateways und, wo möglich, auf Steuerungen per NTP oder PTP. Rechnen Sie mit Ankünften in falscher Reihenfolge und stellen Sie sicher, dass Ihr Speicher Schreibvorgänge in die Vergangenheit akzeptiert. Manche Zeitreihendatenbanken verarbeiten verspätete Schreibvorgänge effizient, andere bestrafen sie.
Speicheroptionen für industrielle Zeitreihendaten
| Option | Stärken | Worauf achten |
|---|---|---|
| Kommerzieller Prozess-Historian | Eingebaute Kompression, Qualität, Asset-Modelle, lange Aufbewahrung | Lizenzierung, proprietärer Zugriff |
| Zeitreihendatenbank (InfluxDB, TimescaleDB, QuestDB) | Offene Schnittstellen, SQL- oder Flux-Abfragen, gute Schreibraten | Kompression, Aufbewahrung und Downsampling entwerfen Sie selbst |
| Data Lake (Parquet auf Object Storage) | Günstiger Langzeitspeicher, ideal für ML-Training | Nicht für Echtzeitabfragen |
| Gestufte Kombination | Heiße Daten in der TSDB, kalte Daten im Lake | Zwei Systeme im Betrieb |
Viele Werke kombinieren einen Historian oder eine TSDB für aktuelle Daten mit einem Lake für Langzeitanalysen. Der Artikel zum MQTT-Historian zeigt, wie Sie die Historian-Seite direkt aus MQTT-Topics aufbauen.
Daten vom Broker in den Speicher bringen
Ein Broker ist ein Router, keine Datenbank: Nachrichten fließen durch ihn hindurch, und Retained Messages halten nur den letzten Wert pro Topic. Um Historie zu speichern, betreiben Sie einen Writer, der abonniert und einfügt. Telegraf ist eine gängige Wahl:
[[inputs.mqtt_consumer]]
servers = ["ssl://broker.connectmq.com:8883"]
topics = ["connectmq/plant1/#"]
client_id = "YOUR_CLIENT_ID"
username = "YOUR_USERNAME"
password = "YOUR_PASSWORD"
qos = 1
data_format = "json"
json_time_key = "ts"
json_time_format = "unix_ms"
tag_keys = ["quality"]
[[inputs.mqtt_consumer.topic_parsing]]
topic = "connectmq/+/+/+/+"
tags = "_/site/line/asset/_"
measurement = "_/_/_/_/measurement"
[[outputs.influxdb_v2]]
urls = ["http://localhost:8086"]
token = "YOUR_INFLUX_TOKEN"
organization = "plant"
bucket = "telemetry"
Die topic_parsing-Regel zeigt, warum sich eine saubere Topic-Struktur auszahlt: Folgen Ihre Topics der ISA-95-Hierarchie eines Unified Namespace (UNS), etwa acme/werk-stuttgart/montage/linie-1/zelle-3/temperatur mit den Ebenen Unternehmen → Standort → Bereich → Linie → Zelle, werden diese Ebenen beim Schreiben automatisch zu Tags in der Zeitreihendatenbank. Abfragen wie „alle Temperaturen in Linie 1 des Werks Stuttgart“ funktionieren dann ohne zusätzliche Mapping-Tabelle. Wie Sie einen solchen Namespace aufsetzen, zeigt Sparkplug B und der Unified Namespace.
Mit ConnectMQ als verwaltetem MQTT-Broker ist Telegraf einfach ein weiterer authentifizierter Client über TLS auf Port 8883. Verwenden Sie QoS 1 für das Abonnement des Writers, damit eine kurze Unterbrechung nicht stillschweigend Daten verwirft, und halten Sie die Session des Writers persistent, wenn während eines Neustarts Nachrichten gepuffert werden sollen.
Checkliste für industrielle Zeitreihendaten
- Klassifizieren Sie jedes Signal: analog langsam, analog schnell, diskret oder hochfrequent, und ordnen Sie es seinem Platz im UNS zu.
- Legen Sie Abtastung und Totband pro Klasse fest, basierend auf der Sensorgenauigkeit.
- Ergänzen Sie einen Heartbeat, damit stabile Signale von toten unterscheidbar sind.
- Setzen Sie Zeitstempel an der Quelle in UTC mit Millisekunden.
- Publizieren Sie mit jedem Wert die Qualität.
- Berechnen Sie für kHz-Signale Merkmale an der Edge.
- Wählen Sie heißen und kalten Speicher und definieren Sie Aufbewahrung und Downsampling.
- Testen Sie verspätete und ungeordnete Schreibvorgänge vor dem Go-live.
Die Architektur rund um all das finden Sie im Leitfaden zu Industrial IoT mit MQTT.
Starten Sie mit ConnectMQ
Erstellen Sie einen kostenlosen Serverless-Broker unter app.connectmq.com, ohne Kreditkarte, und verbinden Sie dann Ihr Gateway und einen Telegraf- oder Python-Writer mit Ihren generierten Zugangsdaten. Der Live-Topic-Explorer von ConnectMQ hilft Ihnen, Publish-Raten und Payloads zu prüfen, bevor Sie sich auf eine Abtast- und Kompressionsrichtlinie festlegen.
Häufig gestellte Fragen
Was sind industrielle Zeitreihendaten?
Es sind zeitgestempelte Mess- und Ereignisdaten, die von Industrieanlagen erzeugt werden, etwa Temperaturen, Drücke, Drehzahlen, Zustände und Zähler. Sie werden üblicherweise mit einem Qualitätsflag gespeichert und für Betrieb, Qualität und Analysen lange aufbewahrt.
Was ist ein guter Totbandwert?
Beginnen Sie nahe an der Genauigkeit oder dem Rauschpegel des Sensors für diese Messung und prüfen Sie dann, ob wichtige Prozessänderungen in den gespeicherten Daten noch sichtbar sind. Dokumentieren Sie den Wert pro Tag-Typ und überprüfen Sie ihn, wenn Sensoren gewechselt werden.
Kann ein MQTT-Broker Zeitreihenhistorie speichern?
Nein. Ein Broker routet Nachrichten und hält höchstens den letzten Retained-Wert pro Topic. Speichern Sie die Historie mit einem Subscriber wie Telegraf, einem Historian-Connector oder einem eigenen Writer.
Sollte ich rohe Schwingungsdaten über MQTT streamen?
In der Regel nicht dauerhaft. Berechnen Sie Merkmale wie RMS, Spitzenwert und Bandenergien an der Edge und publizieren Sie diese; Roh-Wellenform-Schnappschüsse senden Sie nur auf Anforderung oder wenn ein Ereignis sie auslöst.
