Guarda mensajes MQTT en PostgreSQL y TimescaleDB con Python

Para guardar mensajes MQTT en PostgreSQL, ejecuta un pequeño consumidor que se suscriba a tus topics, parsee cada payload e inserte filas por lotes, no una transacción por mensaje. Añade TimescaleDB encima y la misma tabla se convierte en una hypertable con particionado por tiempo, compresión y políticas de retención.

Este tutorial construye ese consumidor en Python con paho-mqtt 2.x y psycopg 3: un esquema que funciona en Postgres puro y en Timescale, un hilo escritor que agrupa por tamaño y por tiempo, una sesión persistente para los reinicios y las consultas que demuestran que los datos llegaron.

Arquitectura

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

El callback MQTT debe terminar rápido, porque paho lo ejecuta en el hilo de red. Por eso on_message solo parsea el payload y pone una fila en una cola; un hilo escritor separado vacía la cola y escribe por lotes. Esa separación es todo el diseño:

Si eres nuevo en la librería cliente, empieza por el tutorial de paho-mqtt 2.x.

Esquema para Postgres y TimescaleDB

Arranca un TimescaleDB local (es PostgreSQL normal con la extensión preinstalada):

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

Guarda unas pocas columnas tipadas que consultes a menudo, más el payload completo como jsonb, para que los campos nuevos nunca rompan la ingesta:

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

-- Solo TimescaleDB (sintaxis 2.13+); omítelo en PostgreSQL puro
SELECT create_hypertable('mqtt_messages', by_range('time'));

CREATE INDEX ON mqtt_messages (device_id, time DESC);

En PostgreSQL puro, el índice sobre (device_id, time DESC) basta para "últimos valores por dispositivo" y consultas por rango con volumen moderado; añade particionado nativo por mes cuando la tabla crezca. En versiones antiguas de Timescale usa create_hypertable('mqtt_messages', 'time').

El escritor por lotes

Instala las dependencias con pip install "paho-mqtt>=2" "psycopg[binary]>=3.1" y ejecuta:

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)))   # bloquea si está llena = 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:      # BD caída: conservar el lote y reintentar
            print(f"database unavailable, retrying: {e}")
            conn = None
            time.sleep(2)
        except psycopg.Error as e:                 # datos inválidos: descartar este lote
            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          # el broker guarda la sesión 1 h sin conexión
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()        # primero dejar de recibir...
client.loop_stop()
stop.set()                 # ...y luego dejar que el escritor vacíe la cola
worker.join()

Algunos detalles que conviene entender:

• executemany en psycopg 3 envía el lote de forma eficiente en modo pipeline, mucho más rápido que un execute más commit por mensaje. Para tasas muy altas, cambia a cur.copy("COPY mqtt_messages (time, topic, device_id, payload) FROM STDIN") y write_row().
• Jsonb(data) adapta el dict parseado a jsonb. Conservar el payload en bruto te permite añadir columnas tipadas más adelante y rellenarlas con SQL.
• La cola acotada es tu válvula de seguridad: si Postgres se ralentiza, rows.put() bloquea el hilo MQTT, el cliente deja de leer el socket y el broker retiene los mensajes en lugar de que tu proceso se quede sin memoria.
• El orden de apagado importa: desconecta de MQTT, luego vacía y haz flush, para que nada recibido quede sin escribir.

Reinicios y garantías de entrega

Con clean_start=False, un client ID fijo y una expiración de sesión, el broker encola los mensajes QoS 1 mientras el consumidor está caído y los entrega al reconectar (session present: True en el log). Más información en sesiones persistentes. Los endpoints públicos de prueba compartidos, incluido broker.connectmq.com, no guardan sesiones offline, así que prueba este comportamiento en tu propio broker privado.

Queda un hueco: paho confirma un mensaje QoS 1 en cuanto on_message retorna, antes de que la fila se haya confirmado en la base de datos. Si el proceso se cae, las filas que siguen en la cola se pierden. Cuando eso importa, crea el cliente con manual_ack=True, lleva msg.mid y msg.qos junto a cada fila y llama a client.ack(mid, qos) para cada mensaje, en el orden de recepción, después de conn.commit(). Recuerda que el broker solo envía un número limitado de mensajes sin confirmar a la vez (Receive Maximum o el límite inflight del broker), así que mantén FLUSH_SECS corto o el throughput quedará limitado por esa ventana.

La entrega at-least-once implica que puede haber duplicados tras las reconexiones. Si te importan, publica un número de secuencia por dispositivo, añade un índice único como (device_id, time, (payload->>'seq')) e inserta con ON CONFLICT DO NOTHING. En una hypertable, los índices únicos deben incluir la columna de tiempo. Consulta mensajes duplicados para conocer las causas.

Prueba el pipeline

1. Arranca la base de datos y aplica schema.sql con psql postgresql://mqtt:mqtt@localhost:5432/telemetry -f schema.sql.
2. Ejecuta python mqtt_to_pg.py.
3. Publica datos de prueba, desde el cliente MQTT online o con un bucle de 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. Comprueba las filas:

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 es una función de TimescaleDB; en PostgreSQL puro usa date_trunc('minute', time). Para probar los reinicios, detén el consumidor, publica diez mensajes más, vuelve a arrancarlo y confirma que el recuento sube en diez (en un broker que guarde sesiones).

Publica lecturas de prueba desde tu navegador

Retención y escalado

• Compresión y retención (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');

• Más consumidores: ejecuta varios escritores con una suscripción compartida como $share/pg-writers/testmqtt/pg/+/telemetry, cada uno con su propio client ID; el broker reparte los mensajes entre ellos. Detalles en suscripciones compartidas.
• Dashboards: el data source PostgreSQL de Grafana lee esta tabla directamente; combínalo con el MQTT data source de Grafana para tener un panel en vivo encima.

Un buen pipeline de MQTT a SQL es aburrido por diseño: parsea rápido, encola con un límite, escribe por lotes y deja que la sesión persistente del broker cubra los reinicios. Pruébalo con unos cuantos mensajes publicados y luego con un reinicio, y sabrás exactamente qué les pasa a tus datos en un mal día.

Preguntas frecuentes