MQTT-Nachrichten mit Python als CSV protokollieren

Wenn Sie MQTT-Nachrichten als CSV protokollieren möchten, wollen Sie meist schnell Antworten: Was hat der Sensor über Nacht gesendet, wie oft meldet sich ein Gerät, und sehen die Daten plausibel aus, bevor Sie etwas Größeres bauen? Dieses Tutorial baut einen kleinen Python-Subscriber, der jede Nachricht in eine CSV-Datei schreibt, Dateien nach Tag und Größe rotiert und Dateien erzeugt, die sich sauber in Excel, pandas oder einem Jupyter Notebook öffnen lassen. Sie können ihn in fünf Minuten gegen den kostenlosen öffentlichen MQTT-Broker laufen lassen und später ohne Codeänderung auf einen privaten Broker umstellen.

Warum CSV immer noch ein guter erster Schritt ist

Eine Zeitreihendatenbank ist das richtige Ziel für langfristige Daten, aber nicht immer der richtige erste Schritt. CSV hat einige Eigenschaften, die es ideal für frühe Analysen machen:

Jedes Werkzeug kann es lesen: Tabellenkalkulationen, pandas, R, DuckDB, sogar ein Texteditor.
Außer Python muss nichts installiert oder betrieben werden.
Dateien lassen sich leicht kopieren, einem Kollegen mailen oder an einen Bug-Report anhängen.
Reine Anhänge-Schreibvorgänge sind einfach und robust, sodass ein Absturz höchstens die letzte Zeile kostet.

Die Nachteile sind ebenso klar. CSV hat keine Indizes, keine Typen und keine Nebenläufigkeitskontrolle, wird daher ab einigen hundert Megabyte langsam und unhandlich und eignet sich schlecht, wenn mehrere Prozesse schreiben müssen. Betrachten Sie es als Erfassungsformat für Debugging, die Inbetriebnahme einer Maschine oder Linie und kurze Experimente. Wird der Logger dauerhaft, steigen Sie auf etwas wie MQTT-Daten mit Python in SQLite oder einen echten Zeitreihenspeicher um.

Was Sie benötigen

Python 3.9 oder neuer
paho-mqtt 2.x (pip install "paho-mqtt>=2.0")
Optional: mosquitto_pub zum Senden von Testnachrichten und pandas für die Analyse

Zum Testen verwenden Sie den öffentlichen MQTT-Broker unter broker.connectmq.com, der für Tests und Prototypen gedacht ist, nicht für Produktionsdaten. Er akzeptiert anonyme Verbindungen mit jeder nicht leeren, eindeutigen Client ID. Der öffentliche Bereich ist für alle lesbar, wählen Sie also ein zufälliges Topic-Präfix wie connectmq/demo/csv-4821/ und senden Sie dort niemals echte Anlagen- oder Kundendaten.

Das CSV-Layout entwerfen

Legen Sie vor dem Programmieren die Zeilenform fest. Es gibt zwei gängige Layouts:

Layout	Spalten	Am besten für	Nachteil
Lang (eine Zeile pro Nachricht)	timestamp, topic, qos, retain, payload, value	Gemischte Topics, unbekannte Payload-Formen, Debugging	Für Diagramme ist ein Pivot nötig
Breit (eine Spalte pro Feld)	timestamp, device, temperature, humidity...	Ein einziges bekanntes Payload-Schema	Bricht, sobald ein neues Feld auftaucht

Der Logger unten verwendet das lange Layout, weil es nie Daten verliert: Was immer über die Leitung kommt, landet in der Datei. Außerdem versucht er, eine numerische Spalte value zu extrahieren, entweder aus einem rein numerischen Payload wie 21.4 oder aus einem JSON-Objekt mit dem Schlüssel value. Das macht schnelle Diagramme trivial, ohne sich auf ein festes Schema festzulegen.

Zeitstempel werden in UTC im Format ISO 8601 mit Millisekunden geschrieben. Ortszeit wirkt freundlicher, erzeugt aber rund um die Zeitumstellung doppelte oder fehlende Stunden, und jedes Analysewerkzeug kann UTC bei der Anzeige in Ortszeit umrechnen.

Der Python-CSV-Logger

Speichern Sie dies als mqtt_csv_logger.py:

import csv
import json
import os
import signal
import sys
import uuid
from datetime import datetime, timezone
from pathlib import Path

import paho.mqtt.client as mqtt

BROKER = os.environ.get("MQTT_HOST", "broker.connectmq.com")
PORT = int(os.environ.get("MQTT_PORT", "1883"))
USERNAME = os.environ.get("MQTT_USERNAME")          # None auf dem öffentlichen Broker
PASSWORD = os.environ.get("MQTT_PASSWORD")
CLIENT_ID = os.environ.get("MQTT_CLIENT_ID", f"csv-logger-{uuid.uuid4().hex[:8]}")
TOPICS = os.environ.get("MQTT_TOPICS", "connectmq/demo/csv-4821/#").split(",")
OUT_DIR = Path(os.environ.get("CSV_DIR", "mqtt_logs"))
MAX_BYTES = int(os.environ.get("CSV_MAX_BYTES", str(50 * 1024 * 1024)))
SKIP_RETAINED = os.environ.get("SKIP_RETAINED", "1") == "1"

FIELDS = ["received_at", "topic", "qos", "retain", "value", "payload"]


class RotatingCsvWriter:
    """Schreibt Zeilen in mqtt_YYYY-MM-DD_NNN.csv, rotiert täglich und nach Größe."""

    def __init__(self, directory: Path, max_bytes: int):
        self.directory = directory
        self.max_bytes = max_bytes
        self.directory.mkdir(parents=True, exist_ok=True)
        self._file = None
        self._writer = None
        self._day = None
        self._part = 0

    def _open(self, day: str):
        if self._file:
            self._file.close()
        if day != self._day:
            self._day, self._part = day, 0
        while True:
            path = self.directory / f"mqtt_{day}_{self._part:03d}.csv"
            if not path.exists() or path.stat().st_size < self.max_bytes:
                break
            self._part += 1
        new_file = not path.exists()
        # utf-8-sig fügt ein BOM hinzu, damit Excel UTF-8 erkennt; newline="" verlangt das csv-Modul
        self._file = open(path, "a", newline="", encoding="utf-8-sig" if new_file else "utf-8")
        self._writer = csv.DictWriter(self._file, fieldnames=FIELDS)
        if new_file:
            self._writer.writeheader()
        print(f"Writing to {path}")

    def write(self, row: dict):
        day = row["received_at"][:10]
        if self._file is None or day != self._day or self._file.tell() >= self.max_bytes:
            if self._file is not None and day == self._day:
                self._part += 1
            self._open(day)
        self._writer.writerow(row)
        self._file.flush()

    def close(self):
        if self._file:
            self._file.close()


def extract_value(text: str):
    """Liefert eine Zahl aus '21.4' oder '{"value": 21.4}', sonst einen leeren String."""
    try:
        return float(text)
    except ValueError:
        pass
    try:
        data = json.loads(text)
        if isinstance(data, dict) and isinstance(data.get("value"), (int, float)):
            return data["value"]
    except json.JSONDecodeError:
        pass
    return ""


writer = RotatingCsvWriter(OUT_DIR, MAX_BYTES)


def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        print(f"Connect failed: {reason_code}")
        return
    print(f"Connected as {CLIENT_ID}")
    for topic in TOPICS:
        client.subscribe(topic.strip(), qos=1)


def on_message(client, userdata, msg):
    if SKIP_RETAINED and msg.retain:
        return
    text = msg.payload.decode("utf-8", errors="replace")
    writer.write({
        "received_at": datetime.now(timezone.utc).isoformat(timespec="milliseconds"),
        "topic": msg.topic,
        "qos": msg.qos,
        "retain": int(msg.retain),
        "value": extract_value(text),
        "payload": text,
    })


def on_disconnect(client, userdata, flags, reason_code, properties):
    print(f"Disconnected: {reason_code}, paho will reconnect")


def main():
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID)
    if USERNAME:
        client.username_pw_set(USERNAME, PASSWORD)
    if PORT == 8883:
        client.tls_set()
    client.on_connect = on_connect
    client.on_message = on_message
    client.on_disconnect = on_disconnect
    client.reconnect_delay_set(min_delay=1, max_delay=30)

    def shutdown(*_):
        client.disconnect()
        writer.close()
        sys.exit(0)

    signal.signal(signal.SIGINT, shutdown)
    signal.signal(signal.SIGTERM, shutdown)

    client.connect(BROKER, PORT, keepalive=60)
    client.loop_forever(retry_first_connection=True)


if __name__ == "__main__":
    main()

Einige Designentscheidungen verdienen eine Erklärung:

Ein einziger Thread. loop_forever() führt Callbacks im selben Thread aus, daher braucht der Writer keinen Lock. Wechseln Sie zu loop_start(), ergänzen Sie einen threading.Lock um write().
Flush nach jeder Zeile. Das kostet etwas Durchsatz, bedeutet aber, dass ein Stromausfall höchstens eine Zeile kostet. Bei sehr hohen Nachrichtenraten flushen Sie stattdessen alle N Zeilen.
Retained Messages standardmäßig übersprungen. Bei jedem Reconnect spielt der Broker Retained Messages erneut aus, was sonst bei jedem Netzwerkaussetzer doppelte Zeilen erzeugen würde.
Abonnement mit QoS 1. Mit QoS 1 publizierte Nachrichten werden mindestens einmal zugestellt, ein Reconnect kann also gelegentlich ein Duplikat erzeugen. Für Analysen ist das meist unproblematisch; lesen Sie MQTT QoS-Stufen erklärt, wenn Sie das genauer verstehen müssen.

Starten und Testdaten senden

Starten Sie den Logger:

pip install "paho-mqtt>=2.0" pandas
python mqtt_csv_logger.py

Publizieren Sie in einem zweiten Terminal einige Messwerte:

for i in 1 2 3 4 5; do
  mosquitto_pub -h broker.connectmq.com -p 1883 -i pub-csv-4821 \
    -t connectmq/demo/csv-4821/line1/temp -m "{\"value\": 2$i.5, \"unit\": \"C\"}"
  mosquitto_pub -h broker.connectmq.com -p 1883 -i pub-csv-4821 \
    -t connectmq/demo/csv-4821/line1/status -m "running"
  sleep 1
done

Sie sollten eine Datei wie mqtt_logs/mqtt_2026-10-16_000.csv mit einer Kopfzeile und zehn Zeilen sehen. Temperaturzeilen haben eine Zahl in value; Statuszeilen lassen die Spalte leer, behalten aber den Rohtext in payload. Haben Sie kein physisches Gerät, zeigt der Artikel zum MQTT-Gerätesimulator, wie Sie realistische Datenströme erzeugen.

Die CSV-Datei mit pandas analysieren

Dateien im langen Format lassen sich mit wenigen Zeilen in Diagramme pivotieren:

import glob
import pandas as pd

files = sorted(glob.glob("mqtt_logs/mqtt_*.csv"))
df = pd.concat((pd.read_csv(f, encoding="utf-8-sig") for f in files), ignore_index=True)
df["received_at"] = pd.to_datetime(df["received_at"], utc=True)

numeric = df.dropna(subset=["value"])
wide = numeric.pivot_table(index="received_at", columns="topic", values="value")
print(wide.resample("1min").mean().tail())

# Nachrichtenrate pro Topic, nützlich, um geschwätzige oder stumme Geräte zu erkennen
print(df.groupby("topic").size().sort_values(ascending=False))

Schon die Übersicht der Nachrichtenraten rechtfertigt oft den Logger: Ein Gerät, das zehnmal häufiger publiziert als erwartet, oder ein Topic, das um 03:00 Uhr verstummt ist, fällt sofort auf.

Checkliste zur Fehlerbehebung

Es wird nichts geschrieben. Prüfen Sie den Topic-Filter. connectmq/demo/csv-4821/# passt auf alles unterhalb des Präfixes, ein Tippfehler in einer Ebene passt aber stillschweigend auf nichts. Abonnieren Sie mit mosquitto_sub -v und demselben Filter, um den Datenverkehr zu bestätigen.
Excel zeigt alles in einer Spalte. Manche Regionaleinstellungen erwarten Semikolons, so auch die deutsche. Nutzen Sie Daten, Aus Text/CSV und wählen Sie Komma als Trennzeichen, oder setzen Sie für dieses Publikum delimiter=";" in DictWriter.
Verstümmelte Zeichen in Excel. Das von utf-8-sig geschriebene BOM behebt das bei neuen Dateien. Dateien aus älteren Versionen des Skripts müssen mit ausgewähltem UTF-8 neu importiert werden.
Leerzeilen zwischen den Zeilen unter Windows. Sie haben die Datei ohne newline="" geöffnet. Das Skript oben berücksichtigt das bereits.
Der Logger trennt ständig die Verbindung. Zwei Prozesse nutzen dieselbe Client ID und werfen sich gegenseitig hinaus. Lassen Sie das Skript eine zufällige ID erzeugen oder setzen Sie eine eindeutige MQTT_CLIENT_ID.
Binäre Payloads sehen wie Rauschen aus. Protobuf- oder komprimierte Payloads sind kein Text. Speichern Sie für diese Topics msg.payload.hex() statt des dekodierten Texts.

Auf einen privaten Broker umsteigen

Der öffentliche MQTT-Broker eignet sich perfekt zum Ausprobieren des Loggers, echte Daten gehören aber auf einen privaten Broker mit Zugangsdaten. Mit ConnectMQ erstellen Sie im Portal einen Broker und erhalten einen generierten Benutzernamen, ein Passwort und eine Client ID. Dann starten Sie dasselbe Skript mit Umgebungsvariablen und TLS auf Port 8883:

export MQTT_HOST=YOUR_BROKER_HOST
export MQTT_PORT=8883
export MQTT_USERNAME=YOUR_USERNAME
export MQTT_PASSWORD=YOUR_PASSWORD
export MQTT_CLIENT_ID=YOUR_CLIENT_ID
export MQTT_TOPICS="factory1/#"
python mqtt_csv_logger.py

Da Client IDs bei ConnectMQ an die ausgegebenen Zugangsdaten gebunden sind, verwenden Sie die Client ID aus dem Portal statt einer zufälligen. Folgt Ihre Anlage einem Unified Namespace (UNS) nach ISA-95, abonnieren Sie gezielt den Teilbaum, den Sie untersuchen, etwa acme/werk-muenchen/montage/linie-1/#, statt den gesamten Namespace. Die Topic-Spalte der CSV enthält dann Werk, Bereich, Linie und Zelle, nach denen Sie in pandas direkt gruppieren können. Mit einem verwalteten MQTT-Broker ist der Logger zudem nur ein weiterer Client: Sie müssen kein Mosquitto daneben betreiben, und im Live-Topic-Explorer des Portals sehen Sie, was der Logger empfangen sollte, bevor Sie der Datei vertrauen.

Wann CSV nicht mehr reicht

CSV-Logging ist ein Zwischenschritt. Wechseln Sie, sobald einer dieser Punkte zutrifft: Die Dateien übersteigen, was Excel bequem öffnet, mehr als ein Prozess muss schreiben, Sie brauchen Abfragen über Wochen an Daten, oder andere Personen benötigen Live-Dashboards. Naheliegende nächste Schritte in dieser Reihe sind das Senden von MQTT-Daten an Google Sheets für gemeinsames, leichtgewichtiges Reporting und eine echte Datenbank für alles Langlebige. Einen Überblick über Python-, JavaScript- und weitere Clients für diese Integrationen gibt der Leitfaden zu MQTT-Client-Bibliotheken. Die paho-mqtt-Dokumentation beschreibt jede hier verwendete Client-Option.

Starten Sie mit ConnectMQ

Richten Sie Ihren CSV-Logger in wenigen Minuten auf einen privaten, per TLS gesicherten Broker aus. Registrieren Sie sich unter app.connectmq.com, wählen Sie den kostenlosen Serverless-Plan ohne Kreditkarte, erstellen Sie einen Broker und tragen Sie die generierten Zugangsdaten in die Umgebungsvariablen oben ein. Im kostenlosen Plan erhalten Sie bis zu 25 Verbindungen, mehr als genug für einen Logger und eine Handvoll Geräte, dazu einen MQTT-Client im Browser zum Senden von Testnachrichten.

Häufig gestellte Fragen

Kann ich MQTT-Nachrichten ohne eigenen Code als CSV protokollieren?

Ja, Werkzeuge wie Node-RED oder Telegraf können Dateien schreiben, doch ein kurzes Python-Skript gibt Ihnen volle Kontrolle über Spalten, Rotation und den Umgang mit Retained Messages. Das Skript in diesem Artikel hat weniger als 150 Zeilen und eine einzige Abhängigkeit.

Verliere ich Nachrichten, während der Logger offline ist?

Mit einer Clean Session ja: Nachrichten, die publiziert werden, während der Logger getrennt ist, werden für ihn nicht gespeichert. Eine persistente Session mit QoS 1 lässt den Broker Nachrichten eine Weile puffern, für eine garantierte Historie nutzen Sie aber eine Datenbank, die von einem dauerhaft laufenden Dienst befüllt wird.

Wie groß können die CSV-Dateien werden?

Das Skript rotiert standardmäßig bei 50 MB und beginnt an jedem UTC-Tag eine neue Datei. Passen Sie CSV_MAX_BYTES an das Analysewerkzeug an; Tabellenkalkulationen werden lange vor pandas träge.

Ist es sicher, Daten vom öffentlichen MQTT-Broker zu protokollieren?

Für Testdaten ja. Alles im öffentlichen Bereich von broker.connectmq.com kann jeder lesen, der dasselbe Topic abonniert. Verwenden Sie für echte Messwerte daher einen privaten Broker mit Zugangsdaten.

MQTT-Nachrichten mit Python als CSV protokollieren für schnelle Analysen