Stocker des messages MQTT dans PostgreSQL et TimescaleDB avec Python

Pour stocker des messages MQTT dans PostgreSQL, faites tourner un petit consommateur qui s'abonne à vos topics, parse chaque payload et insère les lignes par lots, et non une transaction par message. Ajoutez TimescaleDB par-dessus et la même table devient une hypertable avec partitionnement temporel, compression et politiques de rétention.

Ce tutoriel construit ce consommateur en Python avec paho-mqtt 2.x et psycopg 3 : un schéma valable pour Postgres pur comme pour Timescale, un thread d'écriture qui regroupe par taille et par durée, une session persistante pour les redémarrages et les requêtes qui prouvent que les données sont bien arrivées.

Architecture

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

Le callback MQTT doit rendre la main rapidement, car paho l'exécute dans le thread réseau. on_message se contente donc de parser le payload et de déposer une ligne dans une file ; un thread d'écriture séparé vide la file et écrit par lots. Toute la conception tient dans cette séparation :

Si vous découvrez la bibliothèque cliente, commencez par le tutoriel paho-mqtt 2.x.

Schéma pour Postgres et TimescaleDB

Lancez un TimescaleDB local (c'est un PostgreSQL classique avec l'extension préinstallée) :

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

Stockez quelques colonnes typées que vous interrogez souvent, plus le payload complet en jsonb, pour que de nouveaux champs ne cassent jamais l'ingestion :

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

-- TimescaleDB uniquement (syntaxe 2.13+) ; à ignorer sur PostgreSQL pur
SELECT create_hypertable('mqtt_messages', by_range('time'));

CREATE INDEX ON mqtt_messages (device_id, time DESC);

Sur PostgreSQL pur, l'index sur (device_id, time DESC) suffit pour « dernières valeurs par appareil » et les requêtes par plage à volume modéré ; ajoutez un partitionnement natif par mois quand la table grossit. Sur les anciennes versions de Timescale, utilisez create_hypertable('mqtt_messages', 'time').

L'écriture par lots

Installez les dépendances avec pip install "paho-mqtt>=2" "psycopg[binary]>=3.1" puis lancez :

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)))   # bloque si pleine = 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:      # base indisponible : garder le lot, réessayer
            print(f"database unavailable, retrying: {e}")
            conn = None
            time.sleep(2)
        except psycopg.Error as e:                 # données invalides : abandonner ce lot
            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          # le broker garde la session 1 h hors ligne
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()        # d'abord arrêter la réception...
client.loop_stop()
stop.set()                 # ...puis laisser le writer vider la file
worker.join()

Quelques détails à bien comprendre :

• executemany dans psycopg 3 envoie le lot efficacement en mode pipeline, bien plus vite qu'un execute plus un commit par message. Pour des débits très élevés, passez à cur.copy("COPY mqtt_messages (time, topic, device_id, payload) FROM STDIN") et write_row().
• Jsonb(data) adapte le dict parsé au type jsonb. Conserver le payload brut permet d'ajouter plus tard des colonnes typées et de les remplir en SQL.
• La file bornée est votre soupape de sécurité : si Postgres ralentit, rows.put() bloque le thread MQTT, le client cesse de lire le socket et le broker garde les messages au lieu que votre processus manque de mémoire.
• L'ordre d'arrêt compte : déconnectez-vous de MQTT, puis videz la file et faites le flush, pour que rien de reçu ne reste non écrit.

Redémarrages et garanties de livraison

Avec clean_start=False, un client ID fixe et une expiration de session, le broker met en file les messages QoS 1 pendant que le consommateur est arrêté et les livre à la reconnexion (session present: True dans le log). Pour en savoir plus : sessions persistantes. Les endpoints publics de test partagés, y compris broker.connectmq.com, ne conservent pas de session hors ligne : testez ce comportement sur votre propre broker privé.

Il reste une faille : paho acquitte un message QoS 1 dès que on_message retourne, avant que la ligne soit commitée. Si le processus plante, les lignes encore dans la file sont perdues. Quand c'est important, créez le client avec manual_ack=True, transportez msg.mid et msg.qos avec chaque ligne et appelez client.ack(mid, qos) pour chaque message, dans l'ordre de réception, après conn.commit(). N'oubliez pas que le broker n'envoie qu'un nombre limité de messages non acquittés à la fois (Receive Maximum ou la limite inflight du broker) : gardez FLUSH_SECS court, sinon cette fenêtre plafonnera le débit.

Une livraison at-least-once implique que des doublons sont possibles après une reconnexion. S'ils posent problème, publiez un numéro de séquence par appareil, ajoutez un index unique comme (device_id, time, (payload->>'seq')) et insérez avec ON CONFLICT DO NOTHING. Sur une hypertable, les index uniques doivent inclure la colonne de temps. Les causes sont détaillées dans messages dupliqués.

Tester le pipeline

1. Démarrez la base et appliquez schema.sql avec psql postgresql://mqtt:mqtt@localhost:5432/telemetry -f schema.sql.
2. Lancez python mqtt_to_pg.py.
3. Publiez des données de test, depuis le client MQTT en ligne ou avec une boucle shell :

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. Vérifiez les lignes :

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 est une fonction TimescaleDB ; sur PostgreSQL pur, utilisez date_trunc('minute', time). Pour tester les redémarrages, arrêtez le consommateur, publiez dix messages de plus, relancez-le et vérifiez que le total augmente de dix (sur un broker qui conserve les sessions).

Publier des mesures de test depuis le navigateur

Rétention et montée en charge

• Compression et rétention (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');

• Plus de consommateurs : lancez plusieurs writers avec un abonnement partagé comme $share/pg-writers/testmqtt/pg/+/telemetry, chacun avec son propre client ID ; le broker répartit les messages entre eux. Détails dans abonnements partagés.
• Dashboards : la source de données PostgreSQL de Grafana lit cette table directement ; combinez-la avec la source de données MQTT de Grafana pour ajouter un panneau en direct.

Un bon pipeline MQTT vers SQL est volontairement ennuyeux : parser vite, mettre en file avec une limite, écrire par lots et laisser la session persistante du broker couvrir les redémarrages. Testez-le avec quelques messages publiés, puis avec un redémarrage, et vous saurez exactement ce qu'il advient de vos données un mauvais jour.

Questions fréquentes