Industrial DataOps: Werksdaten kontextualisieren

Industrial DataOps ist die Disziplin, rohe Daten der Betriebstechnik (OT) in kontextualisierte, vertrauenswürdige Informationen zu verwandeln und sie jedem Consumer bereitzustellen, der sie braucht – kontinuierlich statt als einmaliges Integrationsprojekt. In diesem Artikel erfahren Sie, welche vier Aufgaben eine DataOps-Schicht erfüllt, wo sie in einer MQTT-basierten Architektur sitzt und wie Sie einen kleinen Kontextualisierungsdienst bauen, der bereinigte Daten erneut in einen gemeinsamen Namespace publiziert – den Unified Namespace (UNS) Ihres Werks.

Was Industrial DataOps tatsächlich bedeutet

Der Begriff stammt aus dem IT-DataOps, das DevOps-Ideen (Automatisierung, Versionierung, Monitoring) auf Datenpipelines anwendet. Im Werk ist das Problem ein anderes. Die Daten werden von SPS, Sensoren und Maschinen erzeugt, die für die Steuerung entworfen wurden, nicht für Analysen. Tags fehlen Einheiten, Zeitstempel stammen von unterschiedlichen Uhren, identische Maschinen tragen unterschiedliche Namen, und niemand ist für das Mapping zwischen ihnen verantwortlich.

Ohne DataOps-Schicht löst jeder Consumer diese Probleme selbst. Das Historian-Team schreibt ein Mapping, der MES-Integrator ein anderes, und das Data-Science-Team bereinigt einen CSV-Export von Hand. Jede Korrektur ist lokal und fragil. Industrial DataOps verlagert diese Arbeit in eine verwaltete Schicht zwischen Quellen und Consumern, sodass die Bereinigung einmal erfolgt und überall wiederverwendet wird.

Das praktische Ergebnis ist eine Werksdatenschicht, in der ein Wert mit Asset, Einheit, Zeitstempel, Qualität und Bedeutung ankommt und in der Änderungen an dieser Schicht wie Code geprüft und versioniert werden.

Die vier Aufgaben einer Industrial-DataOps-Schicht

Aufgabe	Was sie leistet	Beispiel
Kontextualisierung	Ordnet Rohwerten Asset-Identität, Ort und Bedeutung zu	Aus `PLC3.DB10.DBW4` wird `berlin/line2/filler1/speed`
Normalisierung	Vereinheitlicht Einheiten, Skalierung, Datentypen und Namen auf einen Standard	Rohwerte 0–27648 skaliert auf 0–200 bpm
Validierung	Prüft Bereiche, Aktualität und Qualität und markiert fehlerhafte Daten	Ein seit 10 Minuten eingefrorener Wert wird als `STALE` markiert
Governance	Verantwortet das Modell, steuert Änderungen, dokumentiert Consumer	Modellversionen, Änderungsprüfungen, Zugriffsregeln

Kontextualisierung

Kontext macht aus einer Zahl eine Information. Ein Temperaturwert muss wissen, zu welcher Maschine er gehört, zu welchem Teil der Maschine, welches Produkt lief und welche Schicht im Dienst war. Mancher Kontext ist statisch (Asset, Ort) und gehört in den Topic-Pfad oder das Asset-Modell. Anderer ist dynamisch (aktueller Auftrag, Rezept, Betriebsart) und wird am besten als eigene Zustands-Topics publiziert, die Consumer mit der Telemetrie verknüpfen können. Der Leitfaden zur IIoT-Datenmodellierung beschreibt, wie Sie diesen Kontext in Typen und Hierarchien strukturieren.

Normalisierung

Normalisierung sorgt dafür, dass Gleichartiges gleich aussieht. Zwei Motoren verschiedener Hersteller sollten beide current in Ampere melden, beide state mit derselben Aufzählung angeben und beide Millisekunden-Epoch-Zeitstempel verwenden. Typische Normalisierungsschritte sind die Skalierung von Roh- auf technische Werte, Einheitenumrechnung, Umbenennung, Enum-Mapping (Herstellerzustandscodes auf eine gemeinsame Menge) und die Ausrichtung der Zeitstempel auf UTC.

Validierung

OT-Daten fallen auf typische Weise aus: Sensoren frieren ein, Werte springen bei einem Drahtbruch an die Grenzen des Analogbereichs, Uhren driften, Gateways spielen gepufferte Daten in falscher Reihenfolge ab. Eine DataOps-Schicht sollte das abfangen, bevor Consumer es sehen. Verdächtige Daten müssen Sie nicht löschen; oft ist es besser, sie zu behalten und mit einem Qualitätsflag zu versehen, damit ein Historian sie speichern und ein Modelltrainer sie herausfiltern kann.

Governance

Governance ist die am wenigsten technische Aufgabe und diejenige, die am häufigsten übersprungen wird. Jemand muss die Asset-Typen verantworten, Änderungen an Topic-Pfaden freigeben und wissen, welche Anwendungen von welchen Daten abhängen. In der Praxis bedeutet das, Modelle und Mappings in der Versionsverwaltung zu halten, Änderungen zu prüfen und ein Changelog zu veröffentlichen, das Consumer lesen können.

Wo DataOps in einer MQTT-Architektur sitzt

In einem MQTT-basierten Werk ist der Broker das Verteilungsrückgrat, und die DataOps-Logik lebt in Clients um ihn herum. Ein gängiges Layout sieht so aus:

PLCs / sensors
     |
Edge gateway (protocol conversion, raw tags)
     |  publishes raw/...
MQTT broker  <---->  DataOps services (context, normalize, validate)
     |  republish to modeled namespace: site/area/line/cell/...
Consumers: historian, MES, dashboards, ML pipelines

Der Roh-Namespace ist nützlich für die Fehlersuche, aber Consumer sollten nur den modellierten Namespace abonnieren. Diese Trennung erlaubt es, ein Mapping zu korrigieren, ohne das Gateway anzufassen, und Consumer hinzuzufügen, ohne das Mapping anzufassen. Der modellierte Namespace ist genau der Unified Namespace: ein Ort zum Publizieren und Auffinden des aktuellen Zustands, strukturiert nach der Werkshierarchie. In der Praxis folgt diese Hierarchie ISA-95 (IEC 62264): Unternehmen → Standort → Bereich → Linie → Zelle wird zur Topic-Struktur auf dem Broker, etwa acme/werk-stuttgart/montage/linie-1/zelle-3/temperatur. Die DataOps-Schicht ist damit der Dienst, der Roh-Tags in diesen UNS einsortiert; wie Sparkplug B dabei hilft, zeigt Sparkplug B und der Unified Namespace.

Eine ehrliche Anmerkung zu den Werkzeugen: Der Broker routet Nachrichten; er führt Ihren Transformationscode nicht aus. In einem ConnectMQ-Setup ist der Kontextualisierungsdienst ein gewöhnlicher MQTT-Client (ein Python-Prozess, ein Node-RED-Flow oder ein kommerzielles DataOps-Produkt), der Roh-Topics abonniert und modellierte Topics publiziert. ConnectMQ stellt den verwalteten MQTT-Broker, die ISA-95-Topic-Erkennung sowie das Dekodieren von Sparkplug B und dessen erneutes Publizieren in ISA-95-Topics bereit, was einen gängigen Normalisierungsschritt ohne eigenen Code abdeckt.

Einen kleinen Kontextualisierungsdienst bauen

Der folgende Python-Dienst abonniert Roh-Tag-Nachrichten, wendet eine Mapping-Datei an, skaliert und validiert die Werte und publiziert sie erneut in einen ISA-95-Namespace. Zum Experimentieren läuft er gegen den öffentlichen MQTT-Broker; dieser ist für Tests und Prototypen gedacht, verwenden Sie dort also nur simulierte Daten.

Zuerst das Mapping, das in der Versionsverwaltung liegen würde:

# mapping.yaml
modelVersion: 1.0.3
tags:
  PLC3.DB10.DBW4:
    path: berlin/line2/filler1/speed
    unit: bpm
    rawMin: 0
    rawMax: 27648
    engMin: 0
    engMax: 200
    validMin: 0
    validMax: 180
  PLC3.DB10.DBW6:
    path: berlin/line2/filler1/temperature
    unit: degC
    rawMin: 0
    rawMax: 27648
    engMin: -20
    engMax: 120
    validMin: 5
    validMax: 90

Dann der Dienst:

import json
import sys
import time
import uuid

import paho.mqtt.client as mqtt
import yaml

BROKER = "broker.connectmq.com"
PREFIX = sys.argv[1] if len(sys.argv) > 1 else "connectmq/demo/" + uuid.uuid4().hex[:8]
STALE_SECONDS = 600

with open("mapping.yaml", encoding="utf-8") as f:
    MAPPING = yaml.safe_load(f)

last_seen = {}  # Tag -> (Rohwert, Zeitpunkt, zu dem dieser Wert zuerst gesehen wurde)


def scale(raw, m):
    span = m["rawMax"] - m["rawMin"]
    return m["engMin"] + (raw - m["rawMin"]) * (m["engMax"] - m["engMin"]) / span


def quality(tag, raw, value, m, now):
    prev = last_seen.get(tag)
    if prev is None or prev[0] != raw:
        last_seen[tag] = (raw, now)
    elif now - prev[1] > STALE_SECONDS:
        return "STALE"
    if not (m["validMin"] <= value <= m["validMax"]):
        return "OUT_OF_RANGE"
    return "GOOD"


def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(f"{PREFIX}/raw/#", qos=1)
    print("listening on", f"{PREFIX}/raw/#")


def on_message(client, userdata, msg):
    tag = msg.topic.split("/raw/", 1)[1]
    m = MAPPING["tags"].get(tag)
    if m is None:
        client.publish(f"{PREFIX}/dataops/unmapped", json.dumps({"tag": tag}), qos=0)
        return
    raw = json.loads(msg.payload)
    now = time.time()
    value = round(scale(raw["v"], m), 2)
    out = {
        "value": value,
        "unit": m["unit"],
        "ts": raw.get("ts", int(now * 1000)),
        "quality": quality(tag, raw["v"], value, m, now),
        "modelVersion": MAPPING["modelVersion"],
    }
    client.publish(f"{PREFIX}/{m['path']}", json.dumps(out), qos=1)


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="dataops-" + uuid.uuid4().hex[:8])
client.on_connect = on_connect
client.on_message = on_message
client.connect(BROKER, 1883, keepalive=60)
client.loop_forever()

Installieren Sie die Abhängigkeiten mit pip install paho-mqtt pyyaml, starten Sie den Dienst mit einem Präfix Ihrer Wahl und füttern Sie ihn mit einem Rohwert:

python dataops.py connectmq/demo/plant42
mosquitto_pub -h broker.connectmq.com -p 1883 -i raw-test-42 \
  -t "connectmq/demo/plant42/raw/PLC3.DB10.DBW4" -m '{"v": 16000, "ts": 1760000000000}'
mosquitto_sub -h broker.connectmq.com -p 1883 -i sub-test-42 -t "connectmq/demo/plant42/berlin/#" -v

Der Subscriber sieht eine skalierte Geschwindigkeit von etwa 115,7 bpm mit Einheit, Zeitstempel, Qualität und Modellversion. Nicht gemappte Tags landen auf einem eigenen Topic, statt stillschweigend zu verschwinden – so entdecken Sie neue Tags, die jemand einer SPS hinzugefügt hat.

Industrial DataOps einführen: eine Checkliste

Beginnen Sie mit dem Consumer, bei dem es am meisten schmerzt. Wählen Sie die Integration, die derzeit die meiste manuelle Bereinigung erfordert, etwa die OEE-Auswertung oder die MES-Anbindung.
Trennen Sie Roh- und modellierten Namespace vom ersten Tag an und weisen Sie Consumer an, nur den modellierten UNS zu nutzen.
Legen Sie Mappings in die Versionsverwaltung und deployen Sie sie wie Code, mit Reviews.
Publizieren Sie Qualität, nicht nur Werte. Überlassen Sie den Consumern, was sie mit veralteten Daten oder Werten außerhalb des Bereichs tun.
Überwachen Sie die DataOps-Schicht selbst. Verfolgen Sie nicht gemappte Tags, Nachrichtenraten und die Anzahl veralteter Werte; publizieren Sie auch diese als Topics.
Machen Sie Verantwortlichkeiten explizit für jeden Asset-Typ und jeden Standort.
Dokumentieren Sie die Consumer, damit eine Modelländerung daraufhin geprüft werden kann, wer davon abhängt.

Häufige Fehler

DataOps als einmalige Migration behandeln. Werke ändern sich ständig. Wenn niemand das Mapping pflegt, verfällt es innerhalb weniger Monate.
Daten in jedem Dashboard einzeln bereinigen. Logik, die in Visualisierungstools versteckt ist, lässt sich weder wiederverwenden noch testen.
Fehlerhafte Daten verwerfen. Werte zu löschen, vernichtet Beweise. Markieren Sie sie stattdessen.
Die Zeit ignorieren. Ohne Quellzeitstempel und Uhrensynchronisation ist das Verknüpfen von Signalen über Maschinen hinweg Raterei. Industrielle Zeitreihendaten erklärt, warum das für Speicherung und Analyse wichtig ist.

Über ein Werk hinaus skalieren

Wenn mehrere Standorte in dieselbe Struktur publizieren, wird die DataOps-Schicht zum Vertrag zwischen ihnen. Ein Cloud-MQTT-Broker lässt die Dienste jedes Standorts modellierte Daten an einen Ort publizieren, während Kontingente pro Plan und Mandantenisolation die Umgebungen getrennt halten. Für das große Architekturbild zeigt der Leitfaden zu Industrial IoT mit MQTT, wie Gateways, Broker und Consumer zusammenspielen, und der Artikel zum MQTT-Historian behandelt die Speicherseite.

Betreiben Sie Ihre DataOps-Schicht auf ConnectMQ

Erstellen Sie einen kostenlosen Serverless-Broker unter app.connectmq.com, ohne Kreditkarte. Richten Sie Ihr Edge-Gateway auf den Roh-Namespace aus, betreiben Sie Ihren Kontextualisierungsdienst als Client mit YOUR_USERNAME, YOUR_PASSWORD und YOUR_CLIENT_ID, und prüfen Sie mit dem Live-Topic-Explorer und der ISA-95-Geräteansicht von ConnectMQ, ob der modellierte Namespace so aussieht, wie Ihr Modell es vorsieht.

Häufig gestellte Fragen

Was ist Industrial DataOps?

Industrial DataOps ist die Praxis, OT-Daten in einer verwalteten Schicht zu kontextualisieren, zu normalisieren, zu validieren und zu steuern, sodass Historians, MES, Dashboards und Analytics alle dieselben sauberen, gut beschriebenen Informationen erhalten.

Ist Industrial DataOps ein Produkt oder eine Praxis?

Es ist eine Praxis, die durch Produkte unterstützt werden kann. Sie können sie mit kommerziellen DataOps-Plattformen, Node-RED-Flows oder eigenen Diensten umsetzen, solange Mappings versioniert sind, einen Owner haben und angewendet werden, bevor die Daten die Consumer erreichen.

Führt der MQTT-Broker die Transformationen aus?

Nein. Der Broker verteilt Nachrichten; die Transformation läuft in MQTT-Clients, die Roh-Topics abonnieren und modellierte Topics publizieren. ConnectMQ dekodiert zusätzlich Sparkplug B und kann es in ISA-95-Topics erneut publizieren.

Wie hängt Industrial DataOps mit einem Unified Namespace zusammen?

Der Unified Namespace (UNS) ist der Ort, an dem modellierte Daten entlang der ISA-95-Hierarchie publiziert und gefunden werden; DataOps ist die Arbeit, die die Daten darin konsistent und vertrauenswürdig macht. Ein Namespace ohne DataOps wird leicht zu einer gut organisierten Sammlung unsauberer Daten.

Industrial DataOps: Werksdaten kontextualisieren und steuern