Daten mit Python von MQTT zu SQLite zu schicken ist der einfachste Weg, Ihrem IoT-Projekt einen Verlauf zu geben: eine Datei, kein Datenbankserver und volles SQL für Abfragen. In diesem Tutorial bauen Sie einen paho-mqtt-Subscriber, der jede Nachricht von broker.connectmq.com mit Zeitstempel, Topic und Flags speichert, in effizienten Batches schreibt und Reconnects übersteht. Anschließend fragen Sie die Daten nach aktuellen Werten und Stundenmitteln ab und räumen per Aufbewahrungsregel auf.
Warum SQLite für MQTT-Daten?
Ein MQTT-Broker stellt Nachrichten an alle zu, die gerade abonniert haben. Er ist keine Datenbank und hält – abgesehen von einer Retained Message pro Topic – keinen Verlauf vor. Wollen Sie beantworten „Wie hoch war die Temperatur um 3 Uhr nachts?", muss etwas abonnieren und speichern.
SQLite ist für dieses Etwas eine hervorragende erste Wahl:
| Anforderung | Wie SQLite das löst |
|---|---|
| Keine Einrichtung | In der Standardbibliothek von Python enthalten (sqlite3) |
| Eine einzige Datei | Leicht zu sichern, zu kopieren oder in einem GUI-Browser zu öffnen |
| SQL-Abfragen | Filtern, Gruppieren, JSON-Funktionen, Datumsfunktionen |
| Moderate Schreibraten | Tausende Inserts pro Sekunde, wenn sie in Transaktionen gebündelt werden |
| Edge-Geräte | Läuft problemlos auf einem Raspberry Pi oder Industrie-PC |
Es ist nicht das richtige Werkzeug für viele gleichzeitige Schreiber auf verschiedenen Rechnern oder für jahrelange Hochfrequenzdaten einer ganzen Geräteflotte. Für solche Fälle sehen Sie sich den Leitfaden zu MQTT und TimescaleDB oder MQTT-Daten in InfluxDB speichern an. Für einen einzelnen Standort, ein Gateway oder einen Prototyp ist SQLite kaum zu schlagen, etwa als lokaler Puffer auf einem Industrie-PC neben einer Maschine, der Daten auch dann festhält, wenn die Verbindung des Werks zur Cloud ausfällt.
Architektur: den Netzwerk-Thread frei halten
Der naive Ansatz führt in on_message ein INSERT und ein commit() aus. Das funktioniert bei wenigen Nachrichten pro Minute, aber jeder Commit erzwingt eine Synchronisierung auf die Festplatte. Bei höheren Raten gerät der Callback ins Hintertreffen, die Netzwerkschleife von paho stockt, und Keepalive-Pings können sich so lange verzögern, dass der Broker die Verbindung trennt.
Das robuste Design trennt die beiden Aufgaben:
- paho betreibt seine Netzwerkschleife in einem Hintergrund-Thread (
loop_start()). on_messagetut nichts anderes, als die Nachricht in einequeue.Queuezu legen.- Der Haupt-Thread besitzt die SQLite-Verbindung, holt Nachrichten aus der Queue und schreibt sie in Batches, mit einem Commit pro Batch oder einmal pro Sekunde.
Damit umgehen Sie auch die SQLite-Regel, dass eine Verbindung aus dem Thread verwendet werden sollte, der sie erzeugt hat.
Schritt 1: Installieren und vorbereiten
Sie brauchen nur paho-mqtt; sqlite3 ist in Python enthalten:
python -m venv .venv
source .venv/bin/activate # Windows: .venv\Scripts\activate
pip install "paho-mqtt>=2.0"
Wählen Sie ein zufälliges Topic-Präfix, damit sich Ihre Daten nicht mit denen anderer Nutzer des öffentlichen MQTT-Brokers vermischen. Er ist für Tests und Prototypen gedacht, nicht für Produktionsdaten. Dieses Tutorial verwendet connectmq/demo/k7f3q2.
Schritt 2: Der Logger von MQTT zu SQLite in Python
Speichern Sie dies als mqtt_sqlite_logger.py:
import queue
import signal
import sqlite3
import time
import uuid
from datetime import datetime, timezone
import paho.mqtt.client as mqtt
BROKER = "broker.connectmq.com"
PORT = 1883
TOPIC = "connectmq/demo/k7f3q2/#"
DB_PATH = "mqtt_messages.db"
BATCH_SIZE = 200
FLUSH_INTERVAL = 1.0 # Sekunden
SCHEMA = """
CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY,
received_at TEXT NOT NULL,
topic TEXT NOT NULL,
payload TEXT,
qos INTEGER NOT NULL,
retained INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_messages_topic_time ON messages (topic, received_at);
CREATE INDEX IF NOT EXISTS idx_messages_time ON messages (received_at);
"""
inbox: "queue.Queue[tuple]" = queue.Queue(maxsize=50_000)
running = True
def utc_now() -> str:
# 2026-10-16T08:15:30.123Z, sortierbar und von den Datumsfunktionen von SQLite verstanden
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
def on_connect(client, userdata, flags, reason_code, properties):
if reason_code.is_failure:
print(f"Connect failed: {reason_code}")
return
print(f"Connected, subscribing to {TOPIC}")
client.subscribe(TOPIC, qos=1)
def on_message(client, userdata, msg):
row = (
utc_now(),
msg.topic,
msg.payload.decode("utf-8", errors="replace"),
msg.qos,
int(msg.retain),
)
try:
inbox.put_nowait(row)
except queue.Full:
print("Queue full, dropping message") # die Datenbank kommt nicht hinterher
def on_disconnect(client, userdata, disconnect_flags, reason_code, properties):
print(f"Disconnected: {reason_code}, paho will reconnect")
def open_db(path: str) -> sqlite3.Connection:
db = sqlite3.connect(path)
db.execute("PRAGMA journal_mode=WAL") # Leser blockieren den Schreiber nicht
db.execute("PRAGMA synchronous=NORMAL") # mit WAL sicher, deutlich schneller
db.executescript(SCHEMA)
return db
def flush(db: sqlite3.Connection, batch: list) -> None:
if not batch:
return
with db: # eine Transaktion pro Batch
db.executemany(
"INSERT INTO messages (received_at, topic, payload, qos, retained) VALUES (?, ?, ?, ?, ?)",
batch,
)
print(f"Stored {len(batch)} messages")
batch.clear()
def stop(*_):
global running
running = False
def main():
signal.signal(signal.SIGINT, stop)
signal.signal(signal.SIGTERM, stop)
db = open_db(DB_PATH)
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"sqlite-{uuid.uuid4().hex[:8]}")
# Privater ConnectMQ-Broker:
# client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD")
# client.tls_set(); und Port 8883 verwenden
client.on_connect = on_connect
client.on_message = on_message
client.on_disconnect = on_disconnect
client.reconnect_delay_set(min_delay=1, max_delay=60)
client.connect_async(BROKER, PORT, keepalive=60)
client.loop_start()
batch: list = []
last_flush = time.monotonic()
try:
while running:
try:
batch.append(inbox.get(timeout=0.2))
except queue.Empty:
pass
if len(batch) >= BATCH_SIZE or time.monotonic() - last_flush >= FLUSH_INTERVAL:
flush(db, batch)
last_flush = time.monotonic()
finally:
client.disconnect()
client.loop_stop()
while not inbox.empty():
batch.append(inbox.get_nowait())
flush(db, batch)
db.close()
print("Logger stopped cleanly")
if __name__ == "__main__":
main()
Starten Sie ihn mit python mqtt_sqlite_logger.py. Publizieren Sie in einem anderen Terminal einige Testdaten:
for i in $(seq 1 20); do
mosquitto_pub -h broker.connectmq.com -p 1883 -i sqlite-test-pub \
-t "connectmq/demo/k7f3q2/sensors/room$((i % 2 + 1))" \
-m "{\"temperature\": $((20 + RANDOM % 5)).$((RANDOM % 10)), \"humidity\": $((40 + RANDOM % 10))}"
sleep 1
done
Der Logger gibt etwa einmal pro Sekunde „Stored 1 messages" aus. Beenden Sie ihn mit Strg+C, und die verbleibende Queue wird vor dem Beenden geschrieben.
Wie der Logger zuverlässig bleibt
Zeitstempel. Der Logger erfasst den Empfangszeitpunkt der Nachricht in UTC mit Millisekunden. Geräteuhren driften, und manche Geräte haben gar keine, daher ist ein Empfangszeitstempel eine vertrauenswürdige Basis. Enthält Ihr Payload einen eigenen Zeitstempel, speichern Sie beide. Der Leitfaden zu Python, MQTT und JSON zeigt, wie Sie auf der Publisher-Seite einen ergänzen.
WAL-Modus. Write-Ahead-Logging erlaubt einem Dashboard oder der sqlite3-CLI, die Datenbank zu lesen, während der Logger schreibt. Ohne WAL können lange Lesevorgänge Inserts blockieren.
Begrenzte Queue. Stockt die Festplatte, wächst die Queue auf bis zu 50.000 Nachrichten und verwirft dann neue mit einer Meldung, statt den Speicher zu erschöpfen. Passen Sie die Größe an Ihr Datenaufkommen an.
Reconnects. connect_async() plus loop_start() versucht es im Hintergrund weiter, und die Subscription wird in on_connect wiederhergestellt. Nachrichten, die publiziert werden, während der Logger offline ist, gehen verloren, sofern Sie keine persistente Session verwenden. Mit MQTT 5 können Sie ein Session-Expiry-Intervall und clean_start=False setzen, damit der Broker QoS-1-Nachrichten für Sie puffert; der öffentliche MQTT-Broker hält Sessions bis zu 30 Minuten vor. Der Leitfaden zu Reconnects mit paho-mqtt erklärt die Details.
Retained Messages. Bei jedem Reconnect sendet der Broker Retained Messages erneut, daher werden sie mehrfach gespeichert. Mit der Spalte retained können Sie sie aus der Analyse herausfiltern.
Schritt 3: Ihren MQTT-Verlauf abfragen
Öffnen Sie die Datenbank mit der SQLite-CLI (sqlite3 mqtt_messages.db) oder einem beliebigen SQLite-Browser und probieren Sie diese Abfragen aus.
Die letzten 10 Nachrichten eines Topics:
SELECT received_at, payload
FROM messages
WHERE topic = 'connectmq/demo/k7f3q2/sensors/room1'
ORDER BY received_at DESC
LIMIT 10;
Letzter Wert pro Topic, mithilfe des dokumentierten Verhaltens von SQLite für nicht aggregierte Spalten zusammen mit MAX():
SELECT topic, payload, MAX(received_at) AS last_seen
FROM messages
WHERE retained = 0
GROUP BY topic;
Stundenmittel aus JSON-Payloads, mit den eingebauten JSON-Funktionen von SQLite:
SELECT strftime('%Y-%m-%d %H:00', received_at) AS hour,
topic,
ROUND(AVG(json_extract(payload, '$.temperature')), 2) AS avg_temperature,
COUNT(*) AS samples
FROM messages
WHERE topic LIKE 'connectmq/demo/k7f3q2/sensors/%'
AND json_valid(payload)
GROUP BY hour, topic
ORDER BY hour;
Nur die letzten sieben Tage behalten und Speicherplatz zurückgewinnen:
DELETE FROM messages
WHERE received_at < strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-7 days');
VACUUM;
Da received_at in einem festen ISO-8601-Format gespeichert ist, sortiert ein einfacher String-Vergleich korrekt, und der Zeitbereichsfilter kann den Index nutzen. Die Aufbewahrungsabfrage können Sie einmal täglich per Cron oder aus dem Logger selbst ausführen.
Um in Python mit den Daten zu arbeiten, liefert pandas.read_sql_query("SELECT ...", sqlite3.connect("mqtt_messages.db")) einen DataFrame. Das Tutorial zu MQTT und pandas behandelt Analyse und Resampling.
Checkliste zur Fehlerbehebung
- „database is locked": Ein anderer Prozess hält eine Schreibtransaktion. Stellen Sie sicher, dass der WAL-Modus aktiv ist und nur ein Logger in die Datei schreibt.
- Nichts gespeichert: Prüfen Sie, ob der Topic-Filter auf das Topic des Publishers passt; testen Sie mit
mosquitto_sub -h broker.connectmq.com -p 1883 -i check-k7f3q2 -t 'connectmq/demo/k7f3q2/#' -v. json_extractliefert NULL: Der Schlüsselname oder JSON-Pfad weicht ab, oder der Payload ist kein JSON. Sehen Sie sich zuerst einige Rohzeilen an.- Die Datenbank wächst schnell: Ergänzen Sie den Aufbewahrungsjob und erwägen Sie, auf der Publisher-Seite nur geänderte Werte zu senden.
- Der Logger verwirft Nachrichten: Erhöhen Sie
BATCH_SIZE, legen Sie die Datenbank auf schnelleren Speicher oder beschränken Sie die Subscription auf die Topics, die Sie wirklich brauchen.
Vom Prototyp zur Produktion
Für echte Installationen betreiben Sie den Logger als Dienst (systemd unter Linux), damit er nach Neustarts wieder anläuft, und richten ihn auf einen privaten statt auf den öffentlichen MQTT-Broker. Mit ConnectMQ erstellen Sie im Portal einen Broker, kopieren den generierten Benutzernamen, das Passwort und die Client ID und aktivieren TLS auf Port 8883; das Tutorial zu Python, MQTT und TLS zeigt die genauen Einstellungen. Der ConnectMQ-Broker speichert selbst keinen Nachrichtenverlauf, daher ist ein externer Subscriber wie dieser das ehrliche und empfohlene Muster. Brauchen Sie einen Ausgangspunkt für den Client-Code, lesen Sie das Paho-MQTT-Python-Tutorial, und für andere Sprachen die Übersicht über MQTT-Client-Bibliotheken. Die Python-Dokumentation zu sqlite3 behandelt Transaktionen und Verbindungsoptionen ausführlich.
Starten Sie mit ConnectMQ
ConnectMQ ist ein verwalteter MQTT-Broker mit MQTT 3.1.1 und 5.0, TLS und WebSocket sowie einem Live-Topic-Explorer, mit dem Sie prüfen, was Ihre Geräte senden, bevor Sie es protokollieren. Beginnen Sie noch heute mit dem Logging von broker.connectmq.com und wechseln Sie dann auf Ihren eigenen MQTT-Broker in der Cloud. Erstellen Sie ein kostenloses Konto im Serverless-Plan, ohne Kreditkarte.
Häufig gestellte Fragen
Kann SQLite kontinuierliche MQTT-Daten verarbeiten?
Ja, für einen einzelnen Logger bei moderaten Raten. Werden Inserts in Transaktionen gebündelt und der WAL-Modus aktiviert, verkraftet SQLite auch auf bescheidener Hardware einen stetigen Datenstrom vieler Sensoren.
Warum nicht direkt in on_message einfügen?
Ein Commit im Callback blockiert die Netzwerkschleife von paho durch Festplatten-I/O. Unter Last verzögert das Keepalive-Pings und Bestätigungen, daher ist eine Queue mit separatem Schreiber zuverlässiger.
Wie vermeide ich Nachrichtenverluste, während der Logger neu startet?
Verwenden Sie QoS 1 mit einer persistenten MQTT-5-Session (clean_start=False und ein Session-Expiry-Intervall), damit der Broker Nachrichten innerhalb seines Session-Expiry-Limits puffert, während Sie offline sind.
Kann ich binäre Payloads in SQLite speichern?
Ja. Ändern Sie die Spalte payload auf BLOB und fügen Sie msg.payload direkt ein, statt es in Text zu dekodieren.
