Einen Unified Namespace (UNS) mit MQTT aufbauen

Ein Unified Namespace (UNS) ist ein einziger, ereignisgetriebener Ort, an dem jedes System eines Unternehmens seinen aktuellen Zustand publisht – geordnet danach, wo sich etwas im Unternehmen befindet, nicht danach, welche Software es erzeugt hat. Praktisch ist das ein MQTT-Broker plus eine Topic-Hierarchie – meist nach ISA-95: Unternehmen / Standort / Bereich / Linie / Zelle –, in der retained Messages den letzten Wert von allem halten.

Dieses Tutorial zeigt, wie du diese Hierarchie entwirfst, entscheidest, was retained wird, Payload-Konventionen wählst, Sparkplug B einordnest und den Namespace mit ein paar kleinen Skripten testest.

Warum ein UNS statt Punkt-zu-Punkt-Integrationen

Klassische Werke integrieren System für System: Das MES fragt das SCADA ab, das ERP importiert eine CSV aus dem MES, ein Dashboard fragt einen Historian. Jeder neue Konsument bedeutet eine neue Integration, und niemand hat einen Ort, an dem der aktuelle Zustand des Betriebs sichtbar ist.

Ein UNS ersetzt das durch einen Hub:

Produzenten – SPS-Gateways, MES, Qualitätssysteme, ERP-Adapter – publishen in ihren Zweig des Baums. Konsumenten abonnieren, was sie brauchen. Niemand muss wissen, wer sonst noch verbunden ist.

Die ISA-95-Hierarchie entwerfen

ISA-95 liefert ein Vokabular für die physische und organisatorische Struktur eines Werks. Bildest du es direkt auf Topic-Ebenen ab, wird der Namespace für OT- und IT-Leute gleichermaßen lesbar:

Die Geschwindigkeit des Füllers liegt also unter acme/dallas/packaging/line1/filler/speed. Unterhalb der ISA-95-Ebenen ergänzt du einen kleinen Satz Unterbereiche nach Datenart, damit Konsumenten Zustand von Events unterscheiden können:

acme/dallas/packaging/line1/filler/state            "running"       retained
acme/dallas/packaging/line1/filler/speed            {"v":118,...}   retained
acme/dallas/packaging/line1/filler/good-count       {"v":48211,...} retained
acme/dallas/packaging/line1/filler/events/fault     {"code":"E14"}  not retained
acme/dallas/packaging/line1/mes/work-order          {"id":"WO-881"} retained
acme/dallas/packaging/line1/quality/last-check      {"result":"ok"} retained
acme/dallas/packaging/line1/filler/cmd/setpoint     (restricted)    not retained

Regeln, die einen Namespace über Jahre gesund halten:

• Kleinbuchstaben, keine Leerzeichen, nur ASCII, Bindestriche innerhalb einer Ebene. Topics unterscheiden Groß- und Kleinschreibung, gemischte Schreibweise garantiert Duplikate.
• Nach Ort benennen, nicht nach Quelle. line1/filler/speed, niemals kepware3/channel2/tag17. Quellen ändern sich, die Linie bleibt.
• Ebenentiefe innerhalb eines Zweigs fix halten, damit acme/+/packaging/+/filler/state über alle Standorte funktioniert.
• Kommandos trennen unter cmd/ und per ACL schützen; die meisten Produzenten sollten dort nie schreiben dürfen.
• Keine Werte in Topic-Namen (kein .../temp/21.5); der Wert gehört in den Payload.

Die allgemeinen Prinzipien findest du unter Best Practices für Topic-Design. Bevor du dich auf ein Layout festlegst, probiere deine Subscription-Muster im Topic Matcher aus.

Retained State ist das Herz eines UNS

Der Namespace sollte einem Client, der sich vor einer Sekunde verbunden hat, die Frage "Was passiert gerade?" beantworten. Genau das leisten Retained Messages: Der Broker speichert pro Topic die letzte retained Nachricht und liefert sie jedem neuen Subscriber sofort aus.

• Zustand retainen: Werte, Status, aktueller Fertigungsauftrag, letztes Qualitätsergebnis, Sollwerte.
• Events nicht retainen: ausgelöste Störungen, ausgeschleuste Teile, Kommandos. Ein neuer Subscriber soll die Störung von gestern nicht sehen, als wäre sie gerade passiert.
• Veraltete Topics löschen, indem du einen leeren retained Payload publishst, wenn eine Maschine stillgelegt wird – sonst füllt sich der Namespace langsam mit Geistern.
• MQTT-5-Message-Expiry nutzen für Zustände, die ohne Aktualisierung bedeutungslos werden, damit veraltete Werte verschwinden statt zu lügen.

Gib jedem Produzenten ein retained status-Topic und ein Last Will (offline), wie im Modbus-Gateway-Tutorial gezeigt. So können Konsumenten einen stabilen Wert von einer toten Quelle unterscheiden.

Payload-Konventionen

Einigt euch auf eine Payload-Form für Zustände und bleibt dabei. Ein einfaches, selbstbeschreibendes JSON funktioniert gut:

{ "v": 118, "unit": "bpm", "ts": "2026-10-17T08:41:07.214Z", "q": "good" }

v ist der Wert, unit erspart das Raten, ts ist der Quellzeitstempel in UTC, q die Qualität. Statische Metadaten (Beschreibungen, Bereiche, physikalische Einheiten) können in einem retained _meta-Geschwister-Topic liegen, damit der heiße Pfad klein bleibt.

Einen Test-Namespace befüllen und durchsuchen

Mit zwei kleinen paho-mqtt-2.x-Skripten kannst du experimentieren, bevor es einen echten Produzenten gibt. Das erste publisht einen retained Snapshot eines winzigen Werks mit einer Stunde Ablaufzeit:

import json, time
import paho.mqtt.client as mqtt
from paho.mqtt.packettypes import PacketTypes
from paho.mqtt.properties import Properties

ROOT = "testmqtt/uns/acme/dallas/packaging/line1"
STATE = {
    "filler/state":      "running",
    "filler/speed":      {"v": 118, "unit": "bpm"},
    "filler/good-count": {"v": 48211, "unit": "pcs"},
    "capper/state":      "idle",
    "mes/work-order":    {"id": "WO-881", "product": "SKU-330ml"},
}

props = Properties(PacketTypes.PUBLISH)
props.MessageExpiryInterval = 3600          # Sekunden

c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, protocol=mqtt.MQTTv5)
c.connect("broker.connectmq.com", 1883)
c.loop_start()
now = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
for path, value in STATE.items():
    body = value if isinstance(value, str) else json.dumps({**value, "ts": now})
    c.publish(f"{ROOT}/{path}", body, qos=1, retain=True, properties=props).wait_for_publish()
c.loop_stop()
c.disconnect()
print("seeded", len(STATE), "topics")

Das zweite verbindet sich als brandneuer Client, sammelt ein paar Sekunden lang, was der Broker ausliefert, und gibt den Namespace als Baum aus – genau das, was ein neuer Konsument sehen würde:

import time
import paho.mqtt.client as mqtt

FILTER = "testmqtt/uns/acme/#"
seen = {}

def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(FILTER, qos=1)

def on_message(client, userdata, msg):
    seen[msg.topic] = (msg.payload.decode(errors="replace"), msg.retain)

c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, protocol=mqtt.MQTTv5)
c.on_connect, c.on_message = on_connect, on_message
c.connect("broker.connectmq.com", 1883)
c.loop_start()
time.sleep(3)
c.loop_stop()

for topic in sorted(seen):
    payload, retained = seen[topic]
    depth = topic.count("/") - FILTER.count("/") + 1
    print("  " * depth + topic.rsplit("/", 1)[1], "=", payload, "(retained)" if retained else "")

Wenn der Browser jedes befüllte Topic mit (retained) markiert ausgibt, funktioniert deine State-Schicht. Denselben Check kannst du visuell im Online-MQTT-Client machen:

Test-Namespace im Browser durchsuchen

Der gemeinsame Endpoint reicht, um das Layout auszuprobieren. Ein echter Namespace hält den Live-Zustand deines Betriebs, also betreibe ihn auf einem Broker, den du kontrollierst.

Wo Sparkplug B hineinpasst

Sparkplug B ist eine Spezifikation auf MQTT für Industriegeräte: eine feste Topic-Struktur (spBv1.0/<group>/<message type>/<edge node>/<device>), Protobuf-Payloads, Birth/Death-Zertifikate und ein Primary Host, der den Zustand verfolgt. Es löst das Edge-Problem sehr gut, aber seine Topics sind nach Nachrichtentyp und Edge Node geordnet, nicht nach ISA-95.

Das Hybridmodell ist nicht ohne Grund beliebt: Die Geräte bekommen das State-Management von Sparkplug, der Rest des Unternehmens einen lesbaren Namespace. Die Abwägungen beschreibt Sparkplug vs. einfaches MQTT. Daten von OPC-UA-Servern gelangen über eine OPC-UA-zu-MQTT-Bridge in denselben Baum.

Test-Checkliste für deinen Namespace

1. Neuer-Subscriber-Test: Ein neuer Client, der acme/<site>/# abonniert, erhält sofort den aktuellen Zustand jedes Assets (alles retained).
2. Event-Test: Events kommen live an, werden einem neuen Subscriber aber nicht erneut zugestellt.
3. Quellen-Tausch-Test: Ersetze einen Produzenten (simuliertes Gateway → echtes Gateway); die Konsumenten merken nichts.
4. Tote-Quelle-Test: Beende einen Produzenten; sein status springt per Last Will auf offline.
5. ACL-Test: Versuche mit den Zugangsdaten eines Gateways in den Zweig einer anderen Linie zu publishen – das muss abgelehnt werden.
6. Wildcard-Test: Jeder Dashboard-Filter trifft im Topic Matcher genau die beabsichtigten Topics.

Bei einem Unified Namespace geht es weniger um Technik als um Vereinbarung: ein Broker, eine Hierarchie, die alle verstehen, retained State für die Gegenwart und klare Regeln, wer wohin publisht. Starte mit einer Linie, befülle sie, durchsuche sie, mach sie absichtlich kaputt – und lass den Baum erst wachsen, wenn sich der erste Zweig bewährt hat.

Häufig gestellte Fragen