MQTT-Daten in Echtzeit mit Python und pandas analysieren

Die Kombination aus MQTT und pandas bietet Ihnen einen schnellen Weg, Live-IoT-Daten in Python zu analysieren, ohne vorher eine Datenbank aufzusetzen. In diesem Tutorial abonnieren Sie mit paho-mqtt einen Broker, puffern Nachrichten sicher, wandeln sie alle paar Sekunden in DataFrames um und berechnen gleitende Statistiken, minütliche Resamples und einfache Anomalie-Markierungen.

Wann MQTT und pandas die richtige Kombination sind

pandas ist eine Batch-Bibliothek. Sie ist hervorragend darin, auf einer ganzen Tabelle von Zeilen auf einmal zu arbeiten, und schlecht darin, Zeile für Zeile aktualisiert zu werden. MQTT ist das Gegenteil: Nachrichten treffen einzeln, in unregelmäßigen Abständen und in einem Netzwerk-Thread ein, den Sie nicht kontrollieren. Der Trick besteht darin, einen kleinen Puffer dazwischenzuschalten und pandas auf Mikro-Batches arbeiten zu lassen.

Dieses Muster passt gut, wenn Sie einen neuen Datenstrom erkunden, eine Berechnung als Prototyp erstellen wollen, bevor sie in Produktionscode wandert, ein leichtgewichtiges Monitoring auf einem Laptop oder Edge-Gateway betreiben oder periodische Zusammenfassungen für einen Bericht erzeugen möchten. Es ist kein Ersatz für eine Zeitreihendatenbank. Wenn Sie Monate an Historie, mehrere gleichzeitig abfragende Nutzer oder dauerhafte Speicherung brauchen, schreiben Sie den Datenstrom in InfluxDB oder TimescaleDB und nutzen Sie pandas stattdessen auf Abfrageergebnissen.

Architektur: Puffern, Leeren, Analysieren

Das Programm hat zwei Threads mit sehr unterschiedlichen Aufgaben.

1. Der paho-Netzwerk-Thread führt on_message für jede eingehende Nachricht aus. Er muss schnell bleiben, daher dekodiert er nur JSON, extrahiert einige Felder und hängt ein einfaches Dictionary an eine collections.deque an.
2. Der Haupt-Thread wacht alle paar Sekunden auf, leert die Deque, wandelt die neuen Zeilen in einen DataFrame um, hängt sie an ein gleitendes Fenster an und führt die Analyse aus.

deque.append und deque.popleft sind in CPython threadsicher, für diese Übergabe ist also kein expliziter Lock nötig. DataFrames innerhalb von on_message zu bauen ist der häufigste Fehler in MQTT-pandas-Code: Jede Erstellung kostet weit mehr als das Anhängen eines Dicts, und während der Callback läuft, kann paho keine Keep-Alives oder anderen Nachrichten verarbeiten.

Eine Datenquelle vorbereiten

Sie brauchen einen Datenstrom mit mehreren Geräten und etwas Rauschen. Am einfachsten ist der Simulator aus unserem Tutorial zum MQTT-Gerätesimulator, der JSON wie dieses an connectmq/demo/<run-id>/<line>/<device>/telemetry auf dem öffentlichen MQTT-Broker veröffentlicht:

{"device": "sensor-0042", "ts": 1760600000123, "temperature": 21.84, "humidity": 44.2}

Starten Sie ihn, notieren Sie die ausgegebene Run-ID und verwenden Sie diese ID unten. Jeder andere JSON-Datenstrom funktioniert ebenfalls, sofern Sie die Feldnamen anpassen. Wenn Sie neu im Parsen von Payloads sind, behandelt der Leitfaden zur Verarbeitung von MQTT-JSON in Python Dekodierung und Validierung ausführlicher.

Installieren Sie die Abhängigkeiten:

pip install "paho-mqtt>=2.0" "pandas>=2.1"

Das vollständige MQTT-pandas-Skript

Speichern Sie dies als mqtt_pandas.py und setzen Sie RUN_ID auf die Run-ID des Simulators.

import json
import time
import uuid
from collections import deque

import pandas as pd
import paho.mqtt.client as mqtt

BROKER = "broker.connectmq.com"
PORT = 1883
RUN_ID = "3f9a1c2e"  # durch die Run-ID Ihres Simulators ersetzen
TOPIC = f"connectmq/demo/{RUN_ID}/+/+/telemetry"
WINDOW = pd.Timedelta(minutes=15)  # wie viel Historie im Speicher bleibt
ANALYSE_EVERY_S = 10
EXPECTED_INTERVAL_S = 30

inbox: deque = deque(maxlen=100_000)


def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        print(f"connect failed: {reason_code}")
        return
    client.subscribe(TOPIC, qos=0)
    print(f"subscribed to {TOPIC}")


def on_message(client, userdata, msg):
    try:
        data = json.loads(msg.payload)
    except (UnicodeDecodeError, json.JSONDecodeError):
        return  # Nicht-JSON-Payloads ignorieren
    levels = msg.topic.split("/")
    inbox.append({
        "ts": data.get("ts", int(time.time() * 1000)),
        "line": levels[3],
        "device": levels[4],
        "temperature": data.get("temperature"),
        "humidity": data.get("humidity"),
    })


def drain() -> pd.DataFrame:
    rows = []
    while inbox:
        rows.append(inbox.popleft())
    if not rows:
        return pd.DataFrame()
    df = pd.DataFrame(rows)
    df["ts"] = pd.to_datetime(df["ts"], unit="ms", utc=True)
    df["temperature"] = pd.to_numeric(df["temperature"], errors="coerce")
    df["humidity"] = pd.to_numeric(df["humidity"], errors="coerce")
    return df


def analyse(window: pd.DataFrame) -> None:
    now = pd.Timestamp.now(tz="UTC")
    w = window.sort_values("ts")

    # 1. Gleitender 5-Minuten-Mittelwert pro Gerät (zeitbasiertes Fenster)
    rolling = (
        w.groupby("device")
        .rolling("5min", on="ts")["temperature"]
        .mean()
        .groupby(level=0)
        .last()
        .rename("temp_5min_mean")
    )

    # 2. Resample pro Linie und Minute
    per_minute = (
        w.set_index("ts")
        .groupby("line")["temperature"]
        .resample("1min")
        .mean()
        .unstack(level=0)
        .round(2)
    )

    # 3. Robuste Anomalie-Markierung: Abstand zum Gerätemedian in MAD-Einheiten
    med = w.groupby("device")["temperature"].transform("median")
    mad = (w["temperature"] - med).abs().groupby(w["device"]).transform("median")
    w = w.assign(score=(w["temperature"] - med).abs() / (mad * 1.4826 + 1e-9))
    anomalies = w[(w["score"] > 6) & (w["ts"] > now - pd.Timedelta(seconds=ANALYSE_EVERY_S * 2))]

    # 4. Inaktive Geräte: länger als zwei Intervalle nicht gesehen
    last_seen = w.groupby("device")["ts"].max()
    stale = last_seen[last_seen < now - pd.Timedelta(seconds=EXPECTED_INTERVAL_S * 2)]

    print(f"\n=== {now:%H:%M:%S} | rows {len(w)} | devices {w['device'].nunique()} ===")
    print("Per-line means, last 3 minutes:")
    print(per_minute.tail(3).to_string())
    print("Hottest devices by 5-min mean:")
    print(rolling.sort_values(ascending=False).head(5).round(2).to_string())
    if not anomalies.empty:
        print("Anomalies:")
        print(anomalies[["ts", "device", "temperature"]].to_string(index=False))
    if not stale.empty:
        print(f"Stale devices: {', '.join(stale.index[:10])}")


def main():
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"pandas-{uuid.uuid4().hex[:8]}")
    client.on_connect = on_connect
    client.on_message = on_message
    client.connect(BROKER, PORT, keepalive=30)
    client.loop_start()

    window = pd.DataFrame()
    try:
        while True:
            time.sleep(ANALYSE_EVERY_S)
            new = drain()
            if not new.empty:
                window = pd.concat([window, new], ignore_index=True)
            if window.empty:
                continue
            cutoff = pd.Timestamp.now(tz="UTC") - WINDOW
            window = window[window["ts"] >= cutoff].reset_index(drop=True)
            analyse(window)
    except KeyboardInterrupt:
        window.to_csv("mqtt_snapshot.csv", index=False)
        print("saved mqtt_snapshot.csv")
    finally:
        client.loop_stop()
        client.disconnect()


if __name__ == "__main__":
    main()

Führen Sie es mit python mqtt_pandas.py aus. Nach den ersten zehn Sekunden sehen Sie eine Tabelle mit Mittelwerten pro Linie, die fünf heißesten Geräte sowie etwaige Anomalien oder inaktive Sensoren. Drücken Sie Strg+C, um das aktuelle Fenster als CSV-Datei zu speichern.

Die Analyse im Detail

Gleitende Statistiken auf einem zeitbasierten Fenster

rolling("5min", on="ts") verwendet ein zeitbasiertes Fenster statt einer festen Zeilenzahl. Dieser Unterschied ist bei IoT-Daten wichtig: Geräte veröffentlichen mit unterschiedlichen Raten und lassen manchmal Messwerte aus, sodass "die letzten 10 Zeilen" für ein Gerät zehn Sekunden und für ein anderes zehn Minuten bedeuten können. Ein Zeitfenster deckt immer denselben Zeitraum ab. Zeitbasiertes Rolling setzt voraus, dass die Zeitspalte innerhalb jeder Gruppe sortiert ist, deshalb wird das Fenster zuerst nach ts sortiert.

Resampling mit pandas

resample("1min").mean() fasst unregelmäßige Messwerte pro Linie in regelmäßige Ein-Minuten-Intervalle zusammen. Das Ergebnis ist eine breite Tabelle mit einer Spalte pro Linie, bereit zum Plotten oder Exportieren. Verwenden Sie kleingeschriebene Aliase wie "1min" und "10s"; neuere pandas-Versionen markieren den älteren großgeschriebenen Sekunden-Alias als veraltet. Ersetzen Sie mean durch agg(["mean", "min", "max", "count"]), wenn Sie eine reichhaltigere Zusammenfassung möchten, und behalten Sie count als günstigen Indikator für Datenqualität im Blick.

In Industrie-4.0-Umgebungen folgt die Topic-Struktur häufig einem Unified Namespace (UNS) nach ISA-95, etwa acme/werk-stuttgart/montage/linie-1/zelle-3/temperatur. Dann zerlegen Sie das Topic in on_message in Spalten für Werk, Bereich, Linie und Zelle und resamplen auf jeder Ebene, zum Beispiel für ein Energiemonitoring pro Halle oder Kennzahlen pro Linie als Vorstufe einer OEE-Auswertung. Nur die Indizes in levels[...] ändern sich.

Robuste Anomalieerkennung

Ein einfacher z-Score nutzt Mittelwert und Standardabweichung, und beide werden genau durch die Spitzen verzerrt, die Sie finden wollen. Das Skript verwendet stattdessen den Median und die mediane absolute Abweichung (MAD), skaliert mit 1,4826, damit sie bei normalverteilten Daten mit einer Standardabweichung vergleichbar ist. Ein Score über sechs ist ein starkes Signal, dass ein Messwert unplausibel ist. Die fehlerhaften Geräte des Simulators, die gelegentlich um Dutzende Grad springen, fallen sofort auf.

Inaktive Geräte erkennen

Die Last-Seen-Prüfung findet Geräte, die länger als zwei erwartete Intervalle geschwiegen haben. In der Produktion ist das oft wertvoller als jeder Schwellenwert auf den Werten selbst, denn ein toter Sensor meldet überhaupt nichts.

Tipps zu Performance und Speicher

Bei einigen Hundert Nachrichten pro Minute verbraucht dieses Skript sehr wenig CPU. Wächst die Nachrichtenrate auf Tausende pro Sekunde, verlagern Sie die schwere Arbeit in eine Datenbank oder einen Stream-Processor und behalten pandas für Ad-hoc-Analysen.

Einen privaten Broker für echte Daten verwenden

Der öffentliche MQTT-Broker ist perfekt für dieses Tutorial, doch seine Topics sind für jeden lesbar, und er ist nur für Tests und Prototypen gedacht. Für echte Anlagen- oder Gebäudedaten verbinden Sie sich mit einem verwalteten MQTT-Broker mit Zugangsdaten und TLS. Nur die Verbindungszeilen ändern sich:

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="YOUR_CLIENT_ID")
client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD")
client.tls_set()  # Let's-Encrypt-Zertifikat, vom System-Speicher als vertrauenswürdig eingestuft
client.connect("broker.connectmq.com", 8883, keepalive=30)

Bei ConnectMQ erhalten Sie Benutzername, Passwort und Client-ID nach dem Erstellen eines Brokers aus dem Portal. Bevor Sie Ihre Analyse starten, ist der Live-Topic-Explorer im Portal ein schneller Weg zu prüfen, welche Topics tatsächlich ankommen und wie ihre Payloads aussehen.

Um neben Ihrer Live-Ansicht in pandas auch eine langfristige Historie zu behalten, schreiben Sie denselben Datenstrom in eine Datenbank. Unser Leitfaden, wie Sie MQTT-Daten in InfluxDB speichern, verwendet dieselbe Topic-Struktur, sodass Sie beides parallel betreiben können. Für weitere Python-Optionen vergleicht die Übersicht der MQTT-Client-Bibliotheken paho-mqtt mit asynchronen Alternativen, und der pandas User Guide erklärt Fensterfunktionen und Resampling ausführlich.

Starten Sie mit ConnectMQ

Sie können alles in diesem Tutorial sofort gegen broker.connectmq.com ausführen. Wenn Ihre Analyse von Demodaten zu echten Geräten wechselt, bietet Ihnen ConnectMQ einen privaten Broker mit TLS, MQTT 5 und einem Portal zur Prüfung von Live-Topics. Erstellen Sie unter app.connectmq.com einen kostenlosen Serverless-Broker, ohne Kreditkarte. Wenn Sie zuerst die Abwägungen beim Hosting verstehen möchten, lesen Sie unseren Leitfaden zur Wahl eines MQTT-Brokers in der Cloud.

Häufig gestellte Fragen

Kann ich jede MQTT-Nachricht direkt an einen pandas-DataFrame anhängen?

Das ist möglich, aber nicht ratsam. Zeilenweises Anhängen ist langsam und blockiert den Netzwerk-Thread von paho, puffern Sie also einfache Dictionaries und bauen Sie stattdessen aus jedem Batch einen DataFrame.

Wie gehe ich mit Geräten um, die mit unterschiedlichen Raten veröffentlichen?

Verwenden Sie zeitbasierte Fenster wie rolling("5min", on="ts") und resample("1min"). Sie decken einen festen Zeitraum ab, unabhängig davon, wie viele Messwerte jedes Gerät gesendet hat.

Welchen Zeitstempel sollte ich verwenden, den aus der Payload oder die Ankunftszeit?

Bevorzugen Sie den Gerätezeitstempel aus der Payload, wenn die Uhr des Geräts vertrauenswürdig ist, denn er bleibt auch bei verzögerten Nachrichten korrekt. Greifen Sie auf die Ankunftszeit zurück, wenn die Payload keinen enthält.

Funktioniert dieser Ansatz mit asyncio-Clients?

Ja. Bei einem asynchronen Client läuft die Nachrichtenschleife als Coroutine, das Muster bleibt aber dasselbe: Zeilen in einem Puffer sammeln und pandas in festen Abständen auf Batches ausführen.