MQTT-Daten im Jupyter Notebook erkunden: Abonnieren, Plotten und Snapshots speichern

Ein MQTT Jupyter Notebook ist der schnellste Weg, einen Datenstrom zu verstehen, den Sie noch nie gesehen haben: welche Topics existieren, wie die Payloads aussehen und wie sich die Werte über die Zeit verhalten. In diesem Leitfaden verbinden Sie sich aus einem Notebook mit einem Broker, prüfen rohe Nachrichten, zeichnen ein Live-Diagramm, speichern Snapshots für spätere Analysen und fahren alles sauber herunter.

Warum Jupyter für die MQTT-Exploration?

Die meisten MQTT-Werkzeuge beantworten jeweils eine Frage. Ein Kommandozeilen-Subscriber zeigt rohe Nachrichten, ein Desktop-Explorer den Topic-Baum und ein Dashboard Diagramme. Ein Notebook vereint alle drei mit dem gesamten Python-Daten-Stack und hält fest, was Sie ausprobiert haben. Wenn Sie eine bestehende Geräteflotte übernehmen, die Maschinendaten einer neuen Linie im Werk sichten, ein neues Sensormodell erhalten oder einen merkwürdigen Wert debuggen müssen, bringt Sie ein Notebook in einer Sitzung von "Was wird veröffentlicht?" zu "Hier ist ein Diagramm und ein bereinigter Datensatz".

Das entscheidende technische Detail: paho-mqtt kann seine Netzwerkschleife mit loop_start() in einem Hintergrund-Thread ausführen. Der Notebook-Kernel bleibt frei, um Zellen auszuführen, während weiter Nachrichten eintreffen. Alles in diesem Leitfaden baut darauf auf.

Einrichtung

Sie benötigen Python 3.9 oder neuer und eine Jupyter-Umgebung: JupyterLab, die klassische Notebook-Oberfläche oder der Notebook-Editor in VS Code verhalten sich für diesen Leitfaden gleich. Installieren Sie die Pakete aus dem Notebook heraus, damit sie in der Umgebung des Kernels landen:

%pip install "paho-mqtt>=2.0" pandas matplotlib pyarrow

pyarrow wird nur für Parquet-Snapshots benötigt. Wenn Sie CSV bevorzugen, können Sie es weglassen.

Zelle 1: Konfiguration

Legen Sie alles, was Sie eventuell ändern möchten, in die erste Zelle. Eine zufällige Run-ID hält Ihre Demo-Topics auf dem öffentlichen MQTT-Broker getrennt, wo jeder alles lesen kann.

import json, time, uuid, threading, random
from collections import deque
from datetime import datetime, timezone

import pandas as pd
import matplotlib.pyplot as plt
import paho.mqtt.client as mqtt

BROKER = "broker.connectmq.com"
PORT = 1883
RUN_ID = uuid.uuid4().hex[:8]
PREFIX = f"connectmq/demo/{RUN_ID}"
SUBSCRIBE_TO = f"{PREFIX}/#"
print("Topic prefix:", PREFIX)

Wenn bereits ein Datenstrom läuft, zum Beispiel aus unserem MQTT-Gerätesimulator, setzen Sie PREFIX stattdessen auf das Präfix dieses Simulators und überspringen Zelle 3.

Zelle 2: Den Subscriber im Hintergrund verbinden

# Erneutes Ausführen dieser Zelle darf keinen alten Client zurücklassen
if "sub" in globals():
    sub.loop_stop()
    sub.disconnect()

messages = deque(maxlen=50_000)

def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        print("connect failed:", reason_code)
    else:
        client.subscribe(SUBSCRIBE_TO, qos=0)

def on_message(client, userdata, msg):
    messages.append({
        "received": datetime.now(timezone.utc),
        "topic": msg.topic,
        "payload": msg.payload,
        "retain": msg.retain,
    })

sub = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"nb-sub-{RUN_ID}")
sub.on_connect = on_connect
sub.on_message = on_message
sub.connect(BROKER, PORT, keepalive=30)
sub.loop_start()

Die Absicherung ganz oben ist die wichtigste Zeile in einem MQTT Jupyter Notebook. Notebook-Zellen werden ständig erneut ausgeführt. Ohne die Absicherung erzeugt jeder Durchlauf einen weiteren Client mit derselben Client-ID, der Broker trennt den vorherigen, um die neueste Session zu bedienen, der Hintergrund-Thread des vorherigen Clients verbindet sich neu und wirft den neuen hinaus, und am Ende kämpfen zwei Clients in einer Endlosschleife. Nachrichten erscheinen doppelt oder gar nicht, und die Ursache ist nicht offensichtlich.

Beachten Sie, dass der Callback rohe Bytes speichert und nichts parst. Triviale Callbacks bedeuten, dass eine fehlerhafte Payload niemals den Netzwerk-Thread zum Absturz bringen kann, und Sie können später entscheiden, wie dekodiert wird.

Zelle 3: Demodaten erzeugen (optional)

Wenn Sie keinen Datenstrom zum Erkunden haben, startet diese Zelle einen kleinen Publisher in einem weiteren Hintergrund-Thread. Er simuliert zwei Minuten lang drei Temperatursensoren und bleibt dabei deutlich unter dem Limit des öffentlichen MQTT-Brokers von 10 Nachrichten pro Sekunde pro Client.

def demo_publisher(duration_s=120):
    pub = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"nb-pub-{RUN_ID}")
    pub.connect(BROKER, PORT, keepalive=30)
    pub.loop_start()
    state = {"room1": 21.0, "room2": 23.5, "server-rack": 30.0}
    end = time.time() + duration_s
    while time.time() < end:
        for device, value in state.items():
            value += random.gauss(0, 0.2)
            state[device] = value
            payload = {"ts": int(time.time() * 1000), "temperature": round(value, 2)}
            pub.publish(f"{PREFIX}/{device}/telemetry", json.dumps(payload))
        time.sleep(1)
    pub.loop_stop()
    pub.disconnect()

threading.Thread(target=demo_publisher, daemon=True).start()

Zelle 4: Rohe Nachrichten prüfen

Bevor Sie Parsing-Code schreiben, sehen Sie sich an, was wirklich ankommt. Payloads aus der Praxis entsprechen häufig nicht dem, was die Dokumentation verspricht.

print(len(messages), "messages so far")
for m in list(messages)[-5:]:
    print(m["topic"], m["payload"][:200])

topics = pd.Series([m["topic"] for m in messages]).value_counts()
topics.head(20)

Die Ausgabe von value_counts() zeigt, wie viele verschiedene Topics existieren und wie aktiv jedes ist. Sie ist oft das erste Anzeichen eines Problems, etwa eines Geräts, das zehnmal häufiger veröffentlicht als seine Geschwister.

Zelle 5: Nachrichten in einen DataFrame umwandeln

def to_frame(msgs):
    rows = []
    for m in msgs:
        try:
            data = json.loads(m["payload"])
        except (UnicodeDecodeError, json.JSONDecodeError):
            continue
        if not isinstance(data, dict):
            continue
        levels = m["topic"].split("/")
        rows.append({
            "ts": pd.to_datetime(data.get("ts"), unit="ms", utc=True) if "ts" in data else m["received"],
            "device": levels[3] if len(levels) > 3 else m["topic"],
            "temperature": data.get("temperature"),
        })
    df = pd.DataFrame(rows)
    if not df.empty:
        df["temperature"] = pd.to_numeric(df["temperature"], errors="coerce")
        df = df.sort_values("ts")
    return df

df = to_frame(list(messages))
df.groupby("device")["temperature"].describe()

Die Funktion geht davon aus, dass der Gerätename auf der vierten Topic-Ebene steht, was zum Demo-Publisher passt. Die Topics des Simulators haben eine zusätzliche Linien-Ebene (<prefix>/<line>/<device>/telemetry), verwenden Sie dort also levels[4] oder, besser, lesen Sie den Gerätenamen aus der Payload, wenn er enthalten ist. Diese eine Zeile an Ihre Topic-Hierarchie anzupassen ist meist die einzige nötige Änderung am Parsing. In Industrie-4.0-Umgebungen folgt diese Hierarchie häufig einem Unified Namespace (UNS) nach ISA-95, etwa acme/werk-stuttgart/montage/linie-1/zelle-3/temperatur. Dann liefert jede Ebene (Enterprise, Site, Area, Line, Cell) eine eigene Spalte, nach der Sie im DataFrame gruppieren können, zum Beispiel alle Zellen einer Linie im Vergleich.

describe() pro Gerät liefert Anzahl, Mittelwert, Streuung und Wertebereich in einer Tabelle. Ausreißer bei min oder max deuten meist auf fehlerhafte Sensoren oder verwechselte Einheiten hin, etwa ein Gerät, das Fahrenheit meldet.

Zelle 6: Live-Daten im Notebook plotten

Das einfachste zuverlässige Live-Diagramm in Jupyter zeichnet eine statische Matplotlib-Abbildung in einer Schleife neu und löscht die vorherige Ausgabe. Es funktioniert in jedem Notebook-Frontend ohne zusätzliche Widgets.

from IPython.display import clear_output

def live_plot(seconds=60, every=2, window="5min"):
    end = time.time() + seconds
    while time.time() < end:
        df = to_frame(list(messages))
        if not df.empty:
            cutoff = df["ts"].max() - pd.Timedelta(window)
            recent = df[df["ts"] >= cutoff]
            fig, ax = plt.subplots(figsize=(10, 4))
            for device, g in recent.groupby("device"):
                ax.plot(g["ts"], g["temperature"], label=device)
            ax.set_title(f"Live temperature, last {window}")
            ax.set_ylabel("deg C")
            ax.legend(loc="upper left")
            clear_output(wait=True)
            plt.show()
        time.sleep(every)

live_plot(seconds=60)

Die Zelle blockiert für die übergebene Dauer, während der Hintergrund-Thread weiter Nachrichten sammelt. Wenn sie fertig ist, bleibt das letzte Diagramm als Aufzeichnung im Notebook. Unterbrechen Sie den Kernel, um früher aufzuhören; der Subscriber läuft weiter, da er in seinem eigenen Thread lebt.

Zelle 7: Snapshots für spätere Analysen speichern

Live-Daten verschwinden, wenn der Kernel neu startet. Speichern Sie einen Snapshot, sobald Sie etwas Interessantes sehen.

stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
df = to_frame(list(messages))
df.to_parquet(f"mqtt_snapshot_{stamp}.parquet", index=False)

raw = pd.DataFrame(list(messages))
raw["payload"] = raw["payload"].apply(lambda b: b.decode("utf-8", errors="replace"))
raw.to_parquet(f"mqtt_raw_{stamp}.parquet", index=False)
print("saved", len(df), "parsed rows and", len(raw), "raw messages")

Dass sowohl der geparste Frame als auch die rohen Nachrichten gespeichert werden, ist Absicht. Wenn Sie später Ihre Parsing-Logik ändern, können Sie sie auf den rohen Snapshot anwenden, statt darauf zu warten, dass dieselbe Situation erneut eintritt.

Zelle 8: Aufräumen

sub.loop_stop()
sub.disconnect()

Führen Sie dies aus, bevor Sie das Notebook schließen. Ein Herunterfahren des Kernels beendet den Thread zwar ebenfalls, doch ein explizites Trennen verhindert, dass auf dem Broker eine halboffene Session zurückbleibt, bis das Keep-Alive abläuft.

Checkliste zur Fehlerbehebung

Vom Notebook in die Produktion

Notebooks sind für die Exploration gedacht. Sobald Sie wissen, welche Felder wichtig sind, verlagern Sie die Logik in ein Skript oder einen Service: Der paho-mqtt-Python-Leitfaden behandelt einen eigenständigen Client, und die MQTT-Analyse mit pandas zeigt das Mikro-Batch-Muster als langlaufendes Programm. Müssen die Daten aufbewahrt werden, ist eine Zeitreihendatenbank plus Grafana meist das richtige Zuhause, wie im Setup des TIG-Stacks beschrieben.

Wechseln Sie für echte Daten, etwa Maschinendaten aus der Produktion, vom öffentlichen MQTT-Broker zu einem privaten Broker. Bei einem verwalteten MQTT-Broker wie ConnectMQ ändern Sie nur die Verbindungszeilen in Zelle 2: sub.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD") aufrufen, sub.tls_set() aufrufen, client_id="YOUR_CLIENT_ID" verwenden und mit Port 8883 verbinden. Das Let's-Encrypt-Zertifikat des Brokers wird vom Standard-Systemspeicher als vertrauenswürdig eingestuft. Um Python-Bibliotheken für diesen Schritt zu vergleichen, siehe den Leitfaden zu MQTT-Client-Bibliotheken; die Jupyter-Dokumentation behandelt Kernels und Umgebungen, falls sich %pip unerwartet verhält.

Starten Sie mit ConnectMQ

Öffnen Sie ein Notebook und richten Sie es auf den öffentlichen MQTT-Broker, um sofort Demodaten zu erkunden. Er ist für Tests und Prototypen gedacht, nicht für Produktionsdaten. Wenn Sie zu echten Sensoren übergehen, bietet Ihnen ConnectMQ einen privaten Broker mit TLS, MQTT 5 und einem Live-Topic-Explorer im Portal, der gut zur Exploration im Notebook passt. Erstellen Sie unter app.connectmq.com einen kostenlosen Serverless-Broker, ohne Kreditkarte.

Häufig gestellte Fragen

Funktioniert paho-mqtt in einem Jupyter Notebook?

Ja. Verwenden Sie loop_start(), damit die Netzwerkschleife in einem Hintergrund-Thread läuft. Vermeiden Sie loop_forever() im Notebook, denn es blockiert die Zelle und den Kernel, bis Sie unterbrechen.

Warum erhalte ich in meinem Notebook jede MQTT-Nachricht doppelt?

Höchstwahrscheinlich haben Sie die Verbindungszelle erneut ausgeführt und einen zweiten Client mit eigenem Abonnement erzeugt, oder zwei Clients kämpfen um dieselbe Client-ID. Stoppen und trennen Sie den alten Client, bevor Sie einen neuen erstellen.

Kann ich einen asyncio-MQTT-Client in Jupyter verwenden?

Ja. Jupyter betreibt bereits eine Event-Loop, sodass Sie einen asynchronen Client direkt in einer Zelle mit await nutzen können. Der Ansatz mit Hintergrund-Thread und paho ist für die interaktive Exploration einfacher.

Welches Format sollte ich für Snapshots verwenden?

Parquet erhält Spaltentypen und Zeitstempel und ist kompakt, was es ideal für spätere Analysen macht. CSV ist portabler, verliert aber Typinformationen.