MQTT-Nachrichten in PostgreSQL und TimescaleDB speichern mit Python

Um MQTT-Nachrichten in PostgreSQL zu speichern, betreibst du einen kleinen Consumer, der deine Topics abonniert, jeden Payload parst und Zeilen in Batches einfügt – nicht eine Transaktion pro Nachricht. Mit TimescaleDB obendrauf wird dieselbe Tabelle zur Hypertable mit Zeitpartitionierung, Kompression und Retention Policies.

Dieses Tutorial baut den Consumer in Python mit paho-mqtt 2.x und psycopg 3: ein Schema, das mit reinem Postgres und mit Timescale funktioniert, ein Writer-Thread, der nach Größe und Zeit bündelt, eine Persistent Session für Neustarts und die Abfragen, die beweisen, dass die Daten angekommen sind.

Architektur

devices ──► MQTT broker ──► Python consumer ──► PostgreSQL / TimescaleDB
                            (on_message → queue → batch writer)

Der MQTT-Callback muss schnell zurückkehren, denn paho führt ihn im Netzwerk-Thread aus. Deshalb parst on_message nur den Payload und legt eine Zeile in eine Queue; ein separater Writer-Thread leert die Queue und schreibt Batches. Diese Trennung ist das ganze Design:

Wenn die Client-Bibliothek neu für dich ist, starte mit dem paho-mqtt-2.x-Tutorial.

Schema für Postgres und TimescaleDB

Starte ein lokales TimescaleDB (das ist normales PostgreSQL mit vorinstallierter Extension):

docker run -d --name tsdb -p 5432:5432 \
  -e POSTGRES_USER=mqtt -e POSTGRES_PASSWORD=mqtt -e POSTGRES_DB=telemetry \
  timescale/timescaledb:latest-pg16

Speichere ein paar typisierte Spalten, die du oft abfragst, plus den kompletten Payload als jsonb, damit neue Felder die Ingestion nie brechen:

CREATE TABLE mqtt_messages (
  time       timestamptz NOT NULL,
  topic      text        NOT NULL,
  device_id  text        NOT NULL,
  payload    jsonb       NOT NULL
);

-- Nur TimescaleDB (Syntax ab 2.13); bei reinem PostgreSQL weglassen
SELECT create_hypertable('mqtt_messages', by_range('time'));

CREATE INDEX ON mqtt_messages (device_id, time DESC);

Bei reinem PostgreSQL reicht der Index auf (device_id, time DESC) für „letzte Werte pro Gerät“ und Zeitbereichsabfragen bei moderater Menge; ergänze native Partitionierung nach Monat, wenn die Tabelle groß wird. Bei älteren Timescale-Versionen nutzt du create_hypertable('mqtt_messages', 'time').

Der Batch-Writer

Installiere die Abhängigkeiten mit pip install "paho-mqtt>=2" "psycopg[binary]>=3.1" und starte:

import json, os, queue, threading, time
from datetime import datetime, timezone

import paho.mqtt.client as mqtt
from paho.mqtt.packettypes import PacketTypes
from paho.mqtt.properties import Properties
import psycopg
from psycopg.types.json import Jsonb

BROKER = os.environ.get("MQTT_HOST", "broker.connectmq.com")
TOPIC = "testmqtt/pg/+/telemetry"
DSN = os.environ.get("PG_DSN", "postgresql://mqtt:mqtt@localhost:5432/telemetry")
BATCH_SIZE, FLUSH_SECS = 500, 1.0
INSERT = "INSERT INTO mqtt_messages (time, topic, device_id, payload) VALUES (%s, %s, %s, %s)"

rows: queue.Queue = queue.Queue(maxsize=50_000)
stop = threading.Event()


def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        print(f"connect failed: {reason_code}")
        return
    print(f"connected, session present: {flags.session_present}")
    client.subscribe(TOPIC, qos=1)


def on_message(client, userdata, msg):
    try:
        data = json.loads(msg.payload)
    except (UnicodeDecodeError, json.JSONDecodeError):
        print(f"skipping non-JSON payload on {msg.topic}")
        return
    ts = data.get("ts") if isinstance(data, dict) else None
    when = (datetime.fromtimestamp(ts / 1000, timezone.utc)
            if isinstance(ts, (int, float)) else datetime.now(timezone.utc))
    device_id = msg.topic.split("/")[2]
    rows.put((when, msg.topic, device_id, Jsonb(data)))   # blockiert, wenn voll = Backpressure


def writer():
    conn, batch = None, []
    while not (stop.is_set() and rows.empty() and not batch):
        deadline = time.monotonic() + FLUSH_SECS
        while len(batch) < BATCH_SIZE and (left := deadline - time.monotonic()) > 0:
            try:
                batch.append(rows.get(timeout=left))
            except queue.Empty:
                break
        if not batch:
            continue
        try:
            if conn is None or conn.closed:
                conn = psycopg.connect(DSN)
            with conn.cursor() as cur:
                cur.executemany(INSERT, batch)
            conn.commit()
            print(f"wrote {len(batch)} rows")
            batch.clear()
        except psycopg.OperationalError as e:      # DB weg: Batch behalten, erneut versuchen
            print(f"database unavailable, retrying: {e}")
            conn = None
            time.sleep(2)
        except psycopg.Error as e:                 # fehlerhafte Daten: Batch verwerfen
            print(f"insert failed, dropping {len(batch)} rows: {e}")
            conn.rollback()
            batch.clear()


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                     client_id="pg-writer-01", protocol=mqtt.MQTTv5)
client.username_pw_set(os.environ["MQTT_USER"], os.environ["MQTT_PASS"])
client.tls_set()
client.on_connect = on_connect
client.on_message = on_message

props = Properties(PacketTypes.CONNECT)
props.SessionExpiryInterval = 3600          # Broker hält die Session 1 h offline
client.connect(BROKER, 8883, clean_start=False, properties=props)

worker = threading.Thread(target=writer)
worker.start()
client.loop_start()
try:
    while not stop.wait(1):
        pass
except KeyboardInterrupt:
    pass
client.disconnect()        # erst den Empfang stoppen ...
client.loop_stop()
stop.set()                 # ... dann den Writer die Queue leeren lassen
worker.join()

Ein paar Details, die du verstehen solltest:

• executemany in psycopg 3 schickt den Batch effizient im Pipeline-Modus – viel schneller als ein execute plus commit pro Nachricht. Für sehr hohe Raten wechselst du zu cur.copy("COPY mqtt_messages (time, topic, device_id, payload) FROM STDIN") und write_row().
• Jsonb(data) adaptiert das geparste Dict auf jsonb. Weil der Roh-Payload erhalten bleibt, kannst du später typisierte Spalten ergänzen und per SQL nachbefüllen.
• Die begrenzte Queue ist dein Sicherheitsventil: Wird Postgres langsam, blockiert rows.put() den MQTT-Thread, der Client liest nicht mehr vom Socket, und der Broker hält die Nachrichten zurück, statt dass deinem Prozess der Speicher ausgeht.
• Die Reihenfolge beim Herunterfahren zählt: erst von MQTT trennen, dann leeren und flushen, damit nichts Empfangenes ungeschrieben bleibt.

Neustarts und Zustellgarantien

Mit clean_start=False, einer festen Client-ID und einem Session-Ablauf puffert der Broker QoS-1-Nachrichten, während der Consumer down ist, und stellt sie beim Reconnect zu (session present: True im Log). Mehr dazu in Persistent Sessions. Geteilte öffentliche Test-Endpunkte, auch broker.connectmq.com, halten keine Offline-Sessions – teste dieses Verhalten auf deinem eigenen privaten Broker.

Eine Lücke bleibt: paho bestätigt eine QoS-1-Nachricht, sobald on_message zurückkehrt – bevor die Zeile committet ist. Stürzt der Prozess ab, sind die Zeilen in der Queue verloren. Wenn das wichtig ist, erstelle den Client mit manual_ack=True, gib msg.mid und msg.qos mit jeder Zeile weiter und rufe nach conn.commit() für jede Nachricht client.ack(mid, qos) auf – in Empfangsreihenfolge. Denk daran, dass der Broker nur eine begrenzte Zahl unbestätigter Nachrichten gleichzeitig schickt (Receive Maximum oder das Inflight-Limit des Brokers); halte FLUSH_SECS also kurz, sonst begrenzt dieses Fenster den Durchsatz.

At-least-once bedeutet: Duplikate sind nach Reconnects möglich. Wenn sie stören, publiziere eine Sequenznummer pro Gerät, lege einen Unique-Index wie (device_id, time, (payload->>'seq')) an und füge mit ON CONFLICT DO NOTHING ein. Auf einer Hypertable müssen Unique-Indizes die Zeitspalte enthalten. Die Ursachen erklärt doppelte Nachrichten.

Die Pipeline testen

1. Datenbank starten und schema.sql mit psql postgresql://mqtt:mqtt@localhost:5432/telemetry -f schema.sql anwenden.
2. python mqtt_to_pg.py starten.
3. Testdaten publizieren – im Online-MQTT-Client oder per Shell-Schleife:

for i in $(seq 1 20); do
  mosquitto_pub -h broker.connectmq.com -p 8883 --capath /etc/ssl/certs \
    -u "$MQTT_USER" -P "$MQTT_PASS" -q 1 -t testmqtt/pg/sensor-0$((i % 3))/telemetry \
    -m "{\"ts\":$(date +%s000),\"seq\":$i,\"temperature\":$((20 + i % 5)).5}"
done

1. Zeilen prüfen:

SELECT device_id, count(*), max(time) FROM mqtt_messages GROUP BY device_id;

SELECT time_bucket('1 minute', time) AS minute, device_id,
       avg((payload->>'temperature')::float) AS avg_temp
FROM mqtt_messages
WHERE time > now() - interval '1 hour'
GROUP BY 1, 2 ORDER BY 1;

time_bucket ist eine TimescaleDB-Funktion; bei reinem PostgreSQL nimmst du date_trunc('minute', time). Um Neustarts zu testen, stoppe den Consumer, publiziere zehn weitere Nachrichten, starte ihn neu und prüfe, dass die Anzahl um zehn steigt (auf einem Broker, der Sessions speichert).

Testwerte aus dem Browser publizieren

Aufbewahrung und Skalierung

• Kompression und Aufbewahrung (TimescaleDB):

ALTER TABLE mqtt_messages SET (timescaledb.compress, timescaledb.compress_segmentby = 'device_id');
SELECT add_compression_policy('mqtt_messages', INTERVAL '7 days');
SELECT add_retention_policy('mqtt_messages', INTERVAL '90 days');

• Mehr Consumer: Betreibe mehrere Writer mit einer Shared Subscription wie $share/pg-writers/testmqtt/pg/+/telemetry, jeden mit eigener Client-ID; der Broker verteilt die Nachrichten auf sie. Details in Shared Subscriptions.
• Dashboards: Die PostgreSQL-Datenquelle von Grafana liest diese Tabelle direkt; kombiniere sie mit der Grafana-MQTT-Datenquelle für ein Live-Panel obendrauf.

Eine gute MQTT-zu-SQL-Pipeline ist absichtlich langweilig: schnell parsen, begrenzt puffern, in Batches schreiben und Neustarts der Persistent Session des Brokers überlassen. Teste sie mit ein paar publizierten Nachrichten und dann mit einem Neustart – dann weißt du genau, was an einem schlechten Tag mit deinen Daten passiert.

Häufig gestellte Fragen