MQTT vers TimescaleDB : stocker des séries temporelles IoT dans PostgreSQL

Envoyer MQTT vers TimescaleDB vous apporte la fiabilité et le SQL de PostgreSQL, avec un stockage et des requêtes conçus pour les séries temporelles. Dans ce tutoriel, vous allez construire un subscriber Python qui regroupe par lots les messages des appareils provenant d'un broker cloud dans une hypertable TimescaleDB, puis ajouter des agrégats continus, de la compression et de la rétention pour que les tableaux de bord restent rapides à mesure que les données augmentent.

Architecture

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

Le broker route les messages ; il n'écrit pas dans les bases de données. Le pont est un client MQTT ordinaire qui s'abonne, analyse et insère. Cette séparation est un atout : vous pouvez redémarrer, faire évoluer ou remplacer le pont sans toucher aux appareils, et le même modèle fonctionne avec n'importe quel broker MQTT managé. ConnectMQ gère la partie MQTT ; TimescaleDB et le pont s'exécutent là où vous hébergez vos données.

Étape 1 : lancer TimescaleDB

Le démarrage le plus rapide passe par l'image Docker officielle :

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

Connectez-vous avec psql -h localhost -U postgres et créez le schéma.

Étape 2 : concevoir l'hypertable

Les appareils IoT publient rarement le même ensemble de champs. Une table étroite, avec une ligne par valeur de métrique, absorbe les nouveaux capteurs sans modification de schéma et se compresse très bien.

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 transforme la table en un ensemble de chunks partitionnés par le temps, un par jour ici. Les requêtes qui filtrent sur le temps ne touchent que les chunks concernés, et les anciens chunks peuvent être compressés ou supprimés en une seule opération peu coûteuse. Choisissez un intervalle de chunk tel que les chunks les plus récents tiennent confortablement en mémoire ; un jour convient à la plupart des flottes petites et moyennes.

Si chaque appareil envoie exactement les mêmes champs, une table large (time, device_id, temperature, humidity, ...) convient aussi et se requête un peu plus simplement. Le pont ci-dessous s'adapte facilement aux deux cas.

Étape 3 : le pont MQTT vers TimescaleDB

Installez les dépendances :

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

Les appareils publient du JSON comme celui-ci sur <prefix>/<device_id>/telemetry :

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

Le pont comporte deux threads. Le thread réseau MQTT se contente d'analyser et de mettre les lignes en file, il ne bloque donc jamais sur la base de données. Un thread d'écriture vide la file et écrit un lot avec COPY de PostgreSQL toutes les secondes ou toutes les 1 000 lignes, selon ce qui arrive en premier.

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):
    """Transforme un message MQTT en lignes (time, device_id, metric, value)."""
    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)  # conserver le lot et réessayer
    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)

    # Session persistante : le broker met en file les messages QoS 1 pendant le redémarrage du pont
    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()

Pourquoi il est construit ainsi

• COPY plutôt que des INSERT ligne par ligne. Le COPY par lots est le moyen le plus rapide de charger de nombreuses lignes dans PostgreSQL et limite le nombre de transactions.
• Une file bornée. Si la base de données reste indisponible longtemps, la mémoire ne croît pas sans limite. Le pont journalise et abandonne au lieu de planter ; dimensionnez MAX_QUEUE en fonction de la coupure à laquelle vous voulez survivre.
• Session MQTT persistante. Avec clean_start=False, un client ID fixe et une expiration de session, le broker conserve les messages QoS 1 en file pendant le redémarrage du pont. Sur le serveur MQTT public gratuit, les sessions durent jusqu'à 30 minutes. Définissez un MQTT_CLIENT_ID stable pour que cela fonctionne ; un ID aléatoire crée une nouvelle session à chaque fois. Contexte dans les sessions persistantes MQTT.
• Horodatages des appareils lorsqu'ils sont disponibles. Les appareils qui mettent en tampon leurs relevés hors ligne les envoient en retard ; utiliser leur propre ts garde la série temporelle correcte.

Étape 4 : tester le pipeline

Démarrez le pont, puis publiez quelques relevés sur le serveur MQTT public gratuit, réservé aux tests. La zone publique est lisible par n'importe qui : utilisez des données synthétiques :

python mqtt_to_timescaledb.py

# dans un autre 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

Vérifiez les données :

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

Étape 5 : des agrégats continus pour des tableaux de bord rapides

Les tableaux de bord qui tracent un mois de données brutes à la seconde sont lents. Un agrégat continu précalcule des cumuls et les rafraîchit de manière incrémentale en arrière-plan :

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 détermine jusqu'où remontent les données tardives prises en compte à chaque rafraîchissement. Si des appareils peuvent rester hors ligne plus de deux heures puis envoyer des relevés mis en tampon, élargissez-le. Interrogez la vue exactement comme une table :

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

Étape 6 : compression et rétention

Les données de capteurs se compressent extrêmement bien lorsqu'elles sont regroupées par appareil et par métrique :

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');

Cela conserve 90 jours de données brutes, compresse tout ce qui a plus d'une semaine et laisse l'agrégat à la minute subsister pour les tendances à long terme. Les versions récentes de TimescaleDB décrivent aussi la compression comme un « columnstore » ; les politiques ci-dessus continuent de fonctionner.

Visualiser les données dans Grafana

La source de données PostgreSQL intégrée à Grafana fonctionne avec TimescaleDB sans modification. Ajoutez une source de données pointant vers la base, activez l'option TimescaleDB dans ses paramètres et connectez-vous avec un rôle en lecture seule plutôt qu'avec le superutilisateur utilisé par le pont. Écrivez les requêtes des panneaux sur l'agrégat continu et laissez les macros de Grafana insérer la plage temporelle du tableau de bord :

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

N'utilisez la table brute readings que pour de courtes plages où le détail à la seconde compte ; tout ce qui couvre plusieurs jours doit interroger l'agrégat. Si vous préférez vous passer de base de données pour une vue en direct rapide, Grafana avec une source de données MQTT peut tracer directement les topics, mais sans conserver d'historique.

Liste de contrôle pour la production

1. Passez du serveur MQTT public gratuit à un broker privé : créez un broker dans le portail ConnectMQ, puis définissez MQTT_USERNAME, MQTT_PASSWORD et MQTT_CLIENT_ID avec les valeurs générées.
2. Conservez TLS sur le port 8883.
3. Exécutez le pont sous systemd ou dans un conteneur avec redémarrage systématique.
4. Utilisez le QoS 1 de bout en bout ; le plan gratuit prend en charge le QoS jusqu'à 1.
5. Rendez les écritures idempotentes si les doublons comptent : le QoS 1 peut redistribuer. Un index unique sur (device_id, metric, time) associé à une table intermédiaire avec ON CONFLICT DO NOTHING règle le problème.
6. Surveillez la profondeur de la file et le nombre de lignes par seconde dans les logs du pont.
7. Pour monter en charge, exécutez plusieurs ponts avec les abonnements partagés MQTT 5 ($share/tsdb/<prefix>/+/telemetry), disponibles à partir des plans ConnectMQ Serverless payants. Consultez les abonnements partagés MQTT.

Dépannage

Si vous préférez commencer avec un stockage plus simple, comparez cette approche avec l'enregistrement des messages MQTT dans MongoDB ou la journalisation des messages MQTT en CSV. Pour d'autres modèles côté client, le guide des bibliothèques clientes MQTT est une bonne lecture complémentaire.

Démarrer avec ConnectMQ

Pointez votre pont TimescaleDB vers un broker privé. Créez un broker Serverless gratuit sur app.connectmq.com sans carte bancaire, définissez les identifiants générés comme variables d'environnement et utilisez l'explorateur de topics en direct du portail pour vérifier ce que publient vos appareils avant que cela n'atteigne la base de données.

Questions fréquentes

ConnectMQ peut-il écrire directement les données MQTT dans TimescaleDB ?

Non, ConnectMQ est le broker MQTT et n'inclut pas d'intégrations avec des bases de données. Exécutez un subscriber comme le pont Python de ce tutoriel, Telegraf ou Node-RED pour écrire les données.

Faut-il une table étroite ou large pour les données MQTT dans TimescaleDB ?

Utilisez une table étroite (time, device, metric, value) lorsque les appareils envoient des champs variables, et une table large lorsque chaque appareil envoie le même ensemble fixe. Les tables étroites ne nécessitent aucune migration lorsque vous ajoutez des capteurs.

Comment éviter de perdre des messages MQTT quand la base de données est indisponible ?

Mettez en tampon dans le pont avec une file bornée et réessayez les écritures, et utilisez une session MQTT persistante avec le QoS 1 pour que le broker conserve les messages pendant le redémarrage du pont.

PostgreSQL seul suffit-il à la place de TimescaleDB ?

Pour de petits volumes, oui. TimescaleDB ajoute le partitionnement automatique, les agrégats continus, la compression et des politiques de rétention qui prennent de la valeur à mesure que les données augmentent.