MQTT-Daten in TimescaleDB: IoT-Zeitreihen in PostgreSQL speichern

Wer MQTT-Daten in TimescaleDB schreibt, erhält die Zuverlässigkeit und das SQL von PostgreSQL zusammen mit Speicherung und Abfragen, die für Zeitreihen gebaut sind. In diesem Tutorial bauen Sie einen Python-Subscriber, der Gerätenachrichten von einem Cloud-Broker gebündelt in eine TimescaleDB-Hypertable schreibt, und ergänzen anschließend Continuous Aggregates, Kompression und Aufbewahrungsregeln, damit Dashboards auch bei wachsenden Datenmengen schnell bleiben. Gerade für Industrie-4.0-Projekte ist die Kombination attraktiv: Maschinendaten, OEE-Kennzahlen und Energiemonitoring landen in derselben PostgreSQL-Welt, die MES- und ERP-Teams ohnehin kennen.

Architektur

devices --MQTT--> broker (ConnectMQ) --MQTT--> Python bridge --COPY--> TimescaleDB --SQL--> Grafana / apps

Der Broker leitet Nachrichten weiter; er schreibt nicht in Datenbanken. Die Bridge ist ein gewöhnlicher MQTT-Client, der abonniert, parst und einfügt. Diese Trennung ist eine Stärke: Sie können die Bridge neu starten, skalieren oder ersetzen, ohne Geräte anzufassen, und dasselbe Muster funktioniert mit jedem verwalteten MQTT-Broker. ConnectMQ übernimmt die MQTT-Seite; TimescaleDB und die Bridge laufen dort, wo Sie Ihre Daten hosten.

Schritt 1: TimescaleDB starten

Am schnellsten geht es mit dem offiziellen Docker-Image:

docker run -d --name tsdb -p 5432:5432 \
  -e POSTGRES_PASSWORD=change-me \
  -v tsdb-data:/var/lib/postgresql/data \
  timescale/timescaledb:latest-pg16

Verbinden Sie sich mit psql -h localhost -U postgres und legen Sie das Schema an.

Schritt 2: Die Hypertable entwerfen

IoT-Geräte publizieren selten dieselben Felder. Eine schmale Tabelle mit einer Zeile pro Messwert nimmt neue Sensoren ohne Schemaänderungen auf und lässt sich sehr gut komprimieren.

CREATE EXTENSION IF NOT EXISTS timescaledb;

CREATE TABLE readings (
    time       TIMESTAMPTZ      NOT NULL,
    device_id  TEXT             NOT NULL,
    metric     TEXT             NOT NULL,
    value      DOUBLE PRECISION NOT NULL
);

SELECT create_hypertable('readings', by_range('time', INTERVAL '1 day'));

CREATE INDEX readings_device_metric_time
    ON readings (device_id, metric, time DESC);

create_hypertable verwandelt die Tabelle in eine Menge zeitlich partitionierter Chunks, hier einen pro Tag. Abfragen mit Zeitfilter berühren nur die relevanten Chunks, und alte Chunks lassen sich in einem einzigen günstigen Vorgang komprimieren oder löschen. Wählen Sie das Chunk-Intervall so, dass die neuesten Chunks bequem in den Arbeitsspeicher passen; ein Tag passt zu den meisten kleinen und mittleren Flotten.

Folgen Ihre Topics einem Unified Namespace (UNS) nach ISA-95, etwa acme/werk-stuttgart/montage/linie-1/zelle-3/telemetry, lohnt es sich, Werk, Bereich und Linie als zusätzliche Spalten oder in einer Stammdatentabelle zu den Geräten zu führen. So können Sie Kennzahlen pro Linie oder Standort direkt in SQL gruppieren.

Sendet jedes Gerät exakt dieselben Felder, ist auch eine breite Tabelle (time, device_id, temperature, humidity, ...) in Ordnung und etwas einfacher abzufragen. Die Bridge unten lässt sich für beide Varianten leicht anpassen.

Schritt 3: Die Bridge für MQTT-Daten in TimescaleDB

Installieren Sie die Abhängigkeiten:

pip install "paho-mqtt>=2,<3" "psycopg[binary]>=3.1"

Geräte publizieren JSON wie dieses an <prefix>/<device_id>/telemetry:

{"ts": 1792137600.5, "temperature": 23.4, "humidity": 41.0, "battery": 3.71}

Die Bridge hat zwei Threads. Der MQTT-Netzwerk-Thread parst nur und reiht Zeilen in eine Queue ein, sodass er nie auf die Datenbank wartet. Ein Writer-Thread leert die Queue und schreibt jede Sekunde oder alle 1.000 Zeilen, je nachdem, was zuerst eintritt, einen Batch per PostgreSQL-COPY.

mqtt_to_timescaledb.py:

import json
import logging
import os
import queue
import signal
import threading
import time
import uuid
from datetime import datetime, timezone

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

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("bridge")

MQTT_HOST = os.getenv("MQTT_HOST", "broker.connectmq.com")
MQTT_PORT = int(os.getenv("MQTT_PORT", "8883"))
MQTT_USER = os.getenv("MQTT_USERNAME")
MQTT_PASS = os.getenv("MQTT_PASSWORD")
CLIENT_ID = os.getenv("MQTT_CLIENT_ID") or f"tsdb-bridge-{uuid.uuid4().hex[:8]}"
PREFIX = os.getenv("MQTT_TOPIC_PREFIX", "connectmq/demo/tsdb-5b7e21")
DSN = os.getenv("PG_DSN", "postgresql://postgres:change-me@localhost:5432/postgres")

BATCH_SIZE = 1000
FLUSH_SECONDS = 1.0
MAX_QUEUE = 100_000

rows: "queue.Queue[tuple]" = queue.Queue(maxsize=MAX_QUEUE)
stop = threading.Event()


def parse(topic: str, payload: bytes):
    """Eine MQTT-Nachricht in Zeilen (time, device_id, metric, value) umwandeln."""
    parts = topic.split("/")
    if len(parts) < 2 or parts[-1] != "telemetry":
        return []
    device_id = parts[-2]
    data = json.loads(payload)
    if not isinstance(data, dict):
        return []
    ts = data.pop("ts", None)
    when = (datetime.fromtimestamp(float(ts), tz=timezone.utc)
            if isinstance(ts, (int, float)) else datetime.now(timezone.utc))
    out = []
    for metric, value in data.items():
        if isinstance(value, bool):
            value = float(value)
        if isinstance(value, (int, float)):
            out.append((when, device_id, metric[:64], float(value)))
    return out


def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        log.error("connect failed: %s", reason_code)
        return
    log.info("connected (session present: %s)", flags.session_present)
    client.subscribe(f"{PREFIX}/+/telemetry", qos=1)


def on_message(client, userdata, msg):
    try:
        for row in parse(msg.topic, msg.payload):
            rows.put_nowait(row)
    except (ValueError, json.JSONDecodeError):
        log.warning("bad payload on %s", msg.topic)
    except queue.Full:
        log.error("queue full, dropping message from %s", msg.topic)


def writer():
    conn = None
    batch = []
    last_flush = time.monotonic()
    while not (stop.is_set() and rows.empty() and not batch):
        try:
            batch.append(rows.get(timeout=0.2))
            if len(batch) < BATCH_SIZE and time.monotonic() - last_flush < FLUSH_SECONDS:
                continue
        except queue.Empty:
            if not batch:
                continue
        try:
            if conn is None or conn.closed:
                conn = psycopg.connect(DSN, autocommit=False)
            with conn.cursor() as cur:
                with cur.copy("COPY readings (time, device_id, metric, value) FROM STDIN") as cp:
                    for r in batch:
                        cp.write_row(r)
            conn.commit()
            log.info("wrote %d rows", len(batch))
            batch.clear()
            last_flush = time.monotonic()
        except psycopg.Error as e:
            log.error("database error, retrying in 5 s: %s", e)
            if conn is not None:
                conn.close()
            conn = None
            time.sleep(5)  # Batch behalten und erneut versuchen
    if conn is not None:
        conn.close()


def main():
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID,
                         protocol=mqtt.MQTTv5)
    if MQTT_USER:
        client.username_pw_set(MQTT_USER, MQTT_PASS)
    if MQTT_PORT == 8883:
        client.tls_set()
    client.on_connect = on_connect
    client.on_message = on_message
    client.reconnect_delay_set(min_delay=1, max_delay=60)

    # Persistente Session: Der Broker puffert QoS-1-Nachrichten, während die Bridge neu startet
    props = Properties(PacketTypes.CONNECT)
    props.SessionExpiryInterval = 1800

    t = threading.Thread(target=writer, daemon=True)
    t.start()

    def shutdown(*_):
        log.info("shutting down")
        client.disconnect()
        stop.set()

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

    client.connect(MQTT_HOST, MQTT_PORT, keepalive=30, clean_start=False, properties=props)
    client.loop_forever(retry_first_connection=True)
    t.join(timeout=30)


if __name__ == "__main__":
    main()

Warum sie so gebaut ist

• COPY statt zeilenweisem INSERT. Gebündeltes COPY ist der schnellste Weg, viele Zeilen in PostgreSQL zu laden, und hält die Zahl der Transaktionen niedrig.
• Eine begrenzte Queue. Ist die Datenbank länger nicht erreichbar, wächst der Speicher nicht unbegrenzt. Die Bridge protokolliert und verwirft, statt abzustürzen; dimensionieren Sie MAX_QUEUE nach dem Ausfall, den Sie überstehen wollen.
• Persistente MQTT-Session. Mit clean_start=False, einer festen Client ID und einem Session-Ablauf hält der Broker gepufferte QoS-1-Nachrichten bereit, während die Bridge neu startet. Auf dem öffentlichen MQTT-Broker bleiben Sessions bis zu 30 Minuten bestehen. Setzen Sie dafür eine stabile MQTT_CLIENT_ID; eine zufällige ID erzeugt jedes Mal eine neue Session. Hintergründe unter MQTT Persistent Sessions.
• Gerätezeitstempel, wenn vorhanden. Geräte, die Messwerte offline puffern, senden sie verspätet; ihr eigenes ts hält die Zeitreihe korrekt.

Schritt 4: Die Pipeline testen

Starten Sie die Bridge und publizieren Sie dann einige Messwerte an den öffentlichen MQTT-Broker. Er ist für Tests und Prototypen gedacht und für jeden lesbar, verwenden Sie also synthetische Daten:

python mqtt_to_timescaledb.py

# in einem anderen Terminal
for i in $(seq 1 20); do
  mosquitto_pub -h broker.connectmq.com -p 1883 -i tsdb-sim-01 -q 1 \
    -t "connectmq/demo/tsdb-5b7e21/sensor-01/telemetry" \
    -m "{\"temperature\": 2$((RANDOM % 10)).$((RANDOM % 10)), \"humidity\": 4$((RANDOM % 10))}"
  sleep 1
done

Prüfen Sie die Daten:

SELECT device_id, metric, count(*), min(time), max(time)
FROM readings GROUP BY device_id, metric;

Schritt 5: Continuous Aggregates für schnelle Dashboards

Dashboards, die einen Monat sekundengenauer Rohdaten plotten, sind langsam. Ein Continuous Aggregate berechnet Verdichtungen vorab und aktualisiert sie inkrementell im Hintergrund:

CREATE MATERIALIZED VIEW readings_1m
WITH (timescaledb.continuous) AS
SELECT time_bucket(INTERVAL '1 minute', time) AS bucket,
       device_id,
       metric,
       avg(value) AS avg_value,
       min(value) AS min_value,
       max(value) AS max_value,
       count(*)   AS samples
FROM readings
GROUP BY bucket, device_id, metric
WITH NO DATA;

SELECT add_continuous_aggregate_policy('readings_1m',
    start_offset      => INTERVAL '2 hours',
    end_offset        => INTERVAL '1 minute',
    schedule_interval => INTERVAL '1 minute');

start_offset legt fest, wie weit zurück verspätete Daten bei jeder Aktualisierung erfasst werden. Können Geräte länger als zwei Stunden offline sein und danach gepufferte Messwerte hochladen, vergrößern Sie den Wert. Fragen Sie die View genau wie eine Tabelle ab:

SELECT bucket, avg_value
FROM readings_1m
WHERE device_id = 'sensor-01' AND metric = 'temperature'
  AND bucket > now() - INTERVAL '24 hours'
ORDER BY bucket;

Schritt 6: Kompression und Aufbewahrung

Sensordaten lassen sich extrem gut komprimieren, wenn sie nach Gerät und Metrik gruppiert sind:

ALTER TABLE readings SET (
    timescaledb.compress,
    timescaledb.compress_segmentby = 'device_id, metric',
    timescaledb.compress_orderby   = 'time DESC'
);

SELECT add_compression_policy('readings', INTERVAL '7 days');
SELECT add_retention_policy('readings', INTERVAL '90 days');

So bleiben 90 Tage Rohdaten erhalten, alles, was älter als eine Woche ist, wird komprimiert, und das Ein-Minuten-Aggregat bleibt für langfristige Trends bestehen. Neuere TimescaleDB-Versionen bezeichnen Kompression auch als "Columnstore"; die obigen Regeln funktionieren weiterhin.

Die Daten in Grafana visualisieren

Die eingebaute PostgreSQL-Datenquelle von Grafana funktioniert unverändert mit TimescaleDB. Fügen Sie eine Datenquelle hinzu, die auf die Datenbank zeigt, aktivieren Sie in deren Einstellungen die TimescaleDB-Option und verbinden Sie sich mit einer Nur-Lese-Rolle statt mit dem Superuser der Bridge. Schreiben Sie Panel-Abfragen gegen das Continuous Aggregate und lassen Sie die Makros von Grafana den Zeitbereich des Dashboards einsetzen:

SELECT bucket AS "time", device_id AS metric, avg_value
FROM readings_1m
WHERE $__timeFilter(bucket) AND metric = 'temperature'
ORDER BY bucket;

Nutzen Sie die Rohdatentabelle readings nur für kurze Zeiträume, in denen sekundengenaue Details zählen; alles, was Tage umfasst, sollte das Aggregat abfragen. Möchten Sie für eine schnelle Live-Ansicht ganz auf die Datenbank verzichten, kann Grafana mit einer MQTT-Datenquelle Topics direkt plotten, speichert aber keine Historie.

Checkliste für die Produktion

1. Wechseln Sie vom öffentlichen MQTT-Broker zu einem privaten Broker: Erstellen Sie im ConnectMQ-Portal einen Broker und setzen Sie MQTT_USERNAME, MQTT_PASSWORD und MQTT_CLIENT_ID auf die generierten Werte.
2. Bleiben Sie bei TLS auf Port 8883.
3. Betreiben Sie die Bridge unter systemd oder als Container mit Restart always.
4. Verwenden Sie durchgängig QoS 1; der kostenlose Plan unterstützt QoS bis 1.
5. Machen Sie Schreibvorgänge idempotent, wenn Duplikate eine Rolle spielen: QoS 1 kann erneut zustellen. Ein eindeutiger Index auf (device_id, metric, time) plus eine Staging-Tabelle mit ON CONFLICT DO NOTHING löst das.
6. Überwachen Sie Queue-Tiefe und Zeilen pro Sekunde in den Logs der Bridge.
7. Zum horizontalen Skalieren betreiben Sie mehrere Bridges mit Shared Subscriptions aus MQTT 5 ($share/tsdb/<prefix>/+/telemetry), verfügbar ab den kostenpflichtigen ConnectMQ Serverless-Plänen. Siehe MQTT Shared Subscriptions.

Fehlerbehebung

Möchten Sie lieber mit einem einfacheren Speicher beginnen, vergleichen Sie diesen Ansatz mit dem Speichern von MQTT-Nachrichten in MongoDB oder dem Protokollieren von MQTT-Nachrichten als CSV. Für weitere Client-Muster ist der Leitfaden zu MQTT-Client-Bibliotheken eine gute nächste Lektüre.

Starten Sie mit ConnectMQ

Richten Sie Ihre TimescaleDB-Bridge auf einen privaten Broker aus. Erstellen Sie unter app.connectmq.com ohne Kreditkarte einen kostenlosen Serverless-Broker, setzen Sie die generierten Zugangsdaten als Umgebungsvariablen und prüfen Sie im Live-Topic-Explorer des Portals, was Ihre Geräte publizieren, bevor es die Datenbank erreicht.

Häufig gestellte Fragen

Kann ConnectMQ MQTT-Daten direkt in TimescaleDB schreiben?

Nein, ConnectMQ ist der MQTT-Broker und enthält keine Datenbankintegrationen. Betreiben Sie einen Subscriber wie die Python-Bridge aus diesem Tutorial, Telegraf oder Node-RED, um die Daten zu schreiben.

Sollte ich für MQTT-Daten in TimescaleDB eine schmale oder breite Tabelle verwenden?

Verwenden Sie eine schmale Tabelle (Zeit, Gerät, Metrik, Wert), wenn Geräte unterschiedliche Felder senden, und eine breite Tabelle, wenn jedes Gerät denselben festen Satz sendet. Schmale Tabellen brauchen keine Migrationen, wenn Sie Sensoren hinzufügen.

Wie vermeide ich den Verlust von MQTT-Nachrichten, wenn die Datenbank ausfällt?

Puffern Sie in der Bridge mit einer begrenzten Queue, wiederholen Sie Schreibvorgänge und nutzen Sie eine persistente MQTT-Session mit QoS 1, damit der Broker Nachrichten vorhält, während die Bridge neu startet.

Reicht einfaches PostgreSQL statt TimescaleDB?

Bei kleinen Datenmengen ja. TimescaleDB ergänzt automatische Partitionierung, Continuous Aggregates, Kompression und Aufbewahrungsregeln, die mit wachsenden Datenmengen wertvoll werden.