MQTT a TimescaleDB: guarda series temporales IoT

Enviar MQTT a TimescaleDB te da la fiabilidad y el SQL de PostgreSQL con un almacenamiento y unas consultas pensados para series temporales. En este tutorial construirás un suscriptor en Python que agrupa en lotes los mensajes de los dispositivos procedentes de un broker en la nube y los escribe en una hypertable de TimescaleDB, y después añadirás agregados continuos, compresión y retención para que los dashboards sigan siendo rápidos a medida que crecen los datos.

Arquitectura

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

El broker enruta mensajes; no escribe en bases de datos. El puente es un cliente MQTT corriente que se suscribe, analiza e inserta. Esa separación es una ventaja: puedes reiniciar, escalar o sustituir el puente sin tocar los dispositivos, y el mismo patrón funciona con cualquier broker MQTT gestionado. ConnectMQ se encarga de la parte MQTT; TimescaleDB y el puente se ejecutan donde alojes tus datos.

Componente	Función	En este tutorial
Dispositivos	Publican telemetría en JSON	Simulados con `mosquitto_pub`
Broker	Enruta mensajes, mantiene sesiones	broker.connectmq.com y después un broker privado
Puente	Se suscribe, valida, agrupa e inserta	Python, paho-mqtt 2.x, psycopg 3
Base de datos	Almacena y agrega series temporales	TimescaleDB sobre PostgreSQL 16

Paso 1: ejecutar TimescaleDB

La forma más rápida de empezar es la imagen oficial de Docker:

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

Conéctate con psql -h localhost -U postgres y crea el esquema.

Paso 2: diseñar la hypertable

Los dispositivos IoT rara vez publican el mismo conjunto de campos. Una tabla estrecha, con una fila por valor de métrica, absorbe sensores nuevos sin cambios de esquema y se comprime muy 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 convierte la tabla en un conjunto de chunks particionados por tiempo, uno por día en este caso. Las consultas que filtran por tiempo solo tocan los chunks relevantes, y los chunks antiguos se pueden comprimir o eliminar en una única operación barata. Elige un intervalo de chunk que permita que los chunks más recientes quepan holgadamente en memoria; un día es adecuado para la mayoría de las flotas pequeñas y medianas.

Si todos los dispositivos envían exactamente los mismos campos, una tabla ancha (time, device_id, temperature, humidity, ...) también es válida y algo más sencilla de consultar. El puente que sigue se adapta fácilmente a cualquiera de los dos casos.

Paso 3: el puente de MQTT a TimescaleDB

Instala las dependencias:

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

Los dispositivos publican un JSON como este en <prefix>/<device_id>/telemetry:

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

El puente tiene dos hilos. El hilo de red MQTT solo analiza y encola filas, así que nunca se bloquea esperando a la base de datos. Un hilo escritor vacía la cola y escribe un lote con COPY de PostgreSQL cada segundo o cada 1.000 filas, lo que ocurra primero.

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):
    """Convierte un mensaje MQTT en filas (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)  # conservar el lote y reintentar
    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)

    # Sesión persistente: el broker encola los mensajes QoS 1 mientras el puente se reinicia
    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()

Por qué está construido así

COPY en lugar de INSERT fila a fila. El COPY por lotes es la forma más rápida de cargar muchas filas en PostgreSQL y mantiene bajo el número de transacciones.
Una cola acotada. Si la base de datos está caída mucho tiempo, la memoria no crece sin límite. El puente registra el error y descarta en lugar de bloquearse; dimensiona MAX_QUEUE según el corte que quieras soportar.
Sesión MQTT persistente. Con clean_start=False, un client ID fijo y una expiración de sesión, el broker conserva los mensajes QoS 1 encolados mientras el puente se reinicia. En el servidor MQTT público gratuito las sesiones duran hasta 30 minutos. Define un MQTT_CLIENT_ID estable para que esto funcione; un ID aleatorio crea una sesión nueva cada vez. Más contexto en sesiones persistentes en MQTT.
Marcas de tiempo del dispositivo cuando existen. Los dispositivos que almacenan lecturas mientras están desconectados las envían tarde; usar su propio ts mantiene correcta la serie temporal.

Paso 4: probar el pipeline

Arranca el puente y publica algunas lecturas en el servidor público. El área pública es legible por cualquiera, así que usa datos sintéticos:

python mqtt_to_timescaledb.py

# en otra 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

Comprueba los datos:

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

Paso 5: agregados continuos para dashboards rápidos

Los dashboards que dibujan un mes de datos en bruto por segundo son lentos. Un agregado continuo precalcula los resúmenes y los actualiza de forma incremental en segundo plano:

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 controla hasta qué punto del pasado se recogen los datos tardíos en cada actualización. Si los dispositivos pueden estar desconectados más de dos horas y después subir lecturas almacenadas, amplíalo. Consulta la vista exactamente igual que una tabla:

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

Paso 6: compresión y retención

Los datos de sensores se comprimen extraordinariamente bien cuando se agrupan por dispositivo y métrica:

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

Así se conservan 90 días de datos en bruto, se comprime todo lo que tenga más de una semana y el agregado de un minuto sigue vivo para las tendencias a largo plazo. Las versiones recientes de TimescaleDB también describen la compresión como "columnstore"; las políticas anteriores siguen funcionando.

Visualizar los datos en Grafana

La fuente de datos PostgreSQL integrada en Grafana funciona con TimescaleDB sin cambios. Añade una fuente de datos que apunte a la base de datos, activa la opción TimescaleDB en su configuración y conéctate con un rol de solo lectura en lugar del superusuario que usa el puente. Escribe las consultas de los paneles contra el agregado continuo y deja que las macros de Grafana inserten el rango de tiempo del dashboard:

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

Usa la tabla readings en bruto solo para rangos cortos en los que importe el detalle por segundo; todo lo que abarque días debería consultar el agregado. Si prefieres prescindir de la base de datos para una vista rápida en vivo, Grafana con una fuente de datos MQTT puede dibujar los topics directamente, pero no guarda historial.

Lista de comprobación para producción

Pasa del broker público a uno privado: crea un broker en el portal de ConnectMQ y define MQTT_USERNAME, MQTT_PASSWORD y MQTT_CLIENT_ID con los valores generados.
Mantén TLS en el puerto 8883.
Ejecuta el puente bajo systemd o como contenedor con restart always.
Usa QoS 1 de extremo a extremo; el plan gratuito admite QoS hasta 1.
Haz que las escrituras sean idempotentes si los duplicados importan: QoS 1 puede reentregar. Un índice único sobre (device_id, metric, time) más una tabla de staging con ON CONFLICT DO NOTHING lo resuelve.
Vigila la profundidad de la cola y las filas por segundo en los logs del puente.
Para escalar horizontalmente, ejecuta varios puentes con suscripciones compartidas de MQTT 5 ($share/tsdb/<prefix>/+/telemetry), disponibles en los planes Serverless de pago de ConnectMQ y superiores. Consulta suscripciones compartidas en MQTT.

Solución de problemas

Síntoma	Causa probable
`relation "readings" does not exist`	El esquema se creó en una base de datos distinta de la que indica `PG_DSN`
El puente conecta pero no escribe nada	El topic no termina en `/telemetry` o el prefijo no coincide
Las filas tienen una hora incorrecta	El reloj del dispositivo está mal; plantéate ignorar los `ts` muy lejanos en el futuro
Huecos tras los reinicios	Client ID aleatorio, así que no hay sesión persistente
El agregado está vacío	`WITH NO DATA` y la política aún no se ha ejecutado; llama a `CALL refresh_continuous_aggregate('readings_1m', NULL, NULL);`

Si prefieres empezar con un almacenamiento más sencillo, compara este enfoque con guardar mensajes MQTT en MongoDB o registrar mensajes MQTT en CSV. Para más patrones de cliente, la guía de bibliotecas cliente MQTT es una buena lectura a continuación.

Empieza con ConnectMQ

Apunta tu puente de TimescaleDB a un broker privado. Crea un broker Serverless gratuito en app.connectmq.com sin tarjeta de crédito, define las credenciales generadas como variables de entorno y usa el explorador de topics en vivo del portal para confirmar lo que publican tus dispositivos antes de que llegue a la base de datos.

Preguntas frecuentes

¿Puede ConnectMQ escribir datos MQTT directamente en TimescaleDB?

No, ConnectMQ es el broker MQTT y no incluye integraciones con bases de datos. Ejecuta un suscriptor como el puente en Python de este tutorial, Telegraf o Node-RED para escribir los datos.

¿Debo usar una tabla estrecha o ancha para los datos MQTT en TimescaleDB?

Usa una tabla estrecha (time, device, metric, value) cuando los dispositivos envían campos variables, y una tabla ancha cuando todos los dispositivos envían el mismo conjunto fijo. Las tablas estrechas no necesitan migraciones cuando añades sensores.

¿Cómo evito perder mensajes MQTT cuando la base de datos está caída?

Almacena en el puente con una cola acotada y reintenta las escrituras, y usa una sesión MQTT persistente con QoS 1 para que el broker retenga los mensajes mientras el puente se reinicia.

¿Basta con PostgreSQL normal en lugar de TimescaleDB?

Para volúmenes pequeños, sí. TimescaleDB añade particionado automático, agregados continuos, compresión y políticas de retención que cobran valor a medida que crecen los datos.

MQTT a TimescaleDB: guardar series temporales IoT en PostgreSQL