Enviar datos de MQTT a SQLite con Python es la forma más sencilla de dar historial a tu proyecto IoT: un solo archivo, sin servidor de base de datos y con SQL completo para las consultas. En este tutorial construirás un suscriptor paho-mqtt que guarda cada mensaje del servidor MQTT público gratuito broker.connectmq.com con su marca de tiempo, topic y flags, escribe en lotes eficientes y sobrevive a las reconexiones. Después consultarás los datos para obtener valores recientes, medias por hora y limpiar los registros antiguos.
¿Por qué SQLite para datos MQTT?
Un broker MQTT entrega los mensajes a quien esté suscrito en ese momento. No es una base de datos y, aparte de un retained message por topic, no guarda historial. Si quieres responder a "¿qué temperatura había a las 3 de la madrugada?", algo tiene que suscribirse y almacenar.
SQLite es una excelente primera opción para ese "algo":
| Necesidad | Cómo la resuelve SQLite |
|---|---|
| Cero configuración | Incluido en la biblioteca estándar de Python (sqlite3) |
| Un solo archivo | Fácil de respaldar, copiar o abrir en un navegador gráfico |
| Consultas SQL | Filtrado, agrupación, funciones JSON y funciones de fecha |
| Tasas de escritura moderadas | Miles de inserciones por segundo si se agrupan en transacciones |
| Dispositivos edge | Funciona sin problemas en una Raspberry Pi o un PC industrial |
No es la herramienta adecuada para muchos escritores concurrentes en máquinas distintas ni para años de datos de alta frecuencia de toda una flota. Para esos casos, consulta la guía de MQTT a TimescaleDB o cómo guardar datos MQTT en InfluxDB. Para un único emplazamiento, un gateway o un prototipo, SQLite es difícil de superar.
Arquitectura: mantén libre el hilo de red
El enfoque ingenuo ejecuta un INSERT y un commit() dentro de on_message. Funciona con unos pocos mensajes por minuto, pero cada commit fuerza una sincronización con el disco. A tasas más altas el callback se queda atrás, el bucle de red de paho se bloquea y los pings de keepalive pueden retrasarse lo suficiente como para que el broker cierre la conexión.
El diseño robusto separa las dos tareas:
- paho ejecuta su bucle de red en un hilo en segundo plano (
loop_start()). on_messagese limita a poner el mensaje en unaqueue.Queue.- El hilo principal es dueño de la conexión SQLite, saca los mensajes de la cola y los escribe por lotes, haciendo commit una vez por lote o una vez por segundo.
Esto también evita la regla de SQLite según la cual una conexión debe usarse desde el hilo que la creó.
Paso 1: instalación y preparación
Solo necesitas paho-mqtt; sqlite3 viene con Python:
python -m venv .venv
source .venv/bin/activate # Windows: .venv\Scripts\activate
pip install "paho-mqtt>=2.0"
Elige un prefijo de topic aleatorio para que tus datos no se mezclen con los de otros usuarios del servidor público. Este tutorial usa connectmq/demo/k7f3q2.
Paso 2: el logger de MQTT a SQLite en Python
Guarda esto como mqtt_sqlite_logger.py:
import queue
import signal
import sqlite3
import time
import uuid
from datetime import datetime, timezone
import paho.mqtt.client as mqtt
BROKER = "broker.connectmq.com"
PORT = 1883
TOPIC = "connectmq/demo/k7f3q2/#"
DB_PATH = "mqtt_messages.db"
BATCH_SIZE = 200
FLUSH_INTERVAL = 1.0 # segundos
SCHEMA = """
CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY,
received_at TEXT NOT NULL,
topic TEXT NOT NULL,
payload TEXT,
qos INTEGER NOT NULL,
retained INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_messages_topic_time ON messages (topic, received_at);
CREATE INDEX IF NOT EXISTS idx_messages_time ON messages (received_at);
"""
inbox: "queue.Queue[tuple]" = queue.Queue(maxsize=50_000)
running = True
def utc_now() -> str:
# 2026-10-16T08:15:30.123Z, ordenable y compatible con las funciones de fecha de SQLite
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
def on_connect(client, userdata, flags, reason_code, properties):
if reason_code.is_failure:
print(f"Connect failed: {reason_code}")
return
print(f"Connected, subscribing to {TOPIC}")
client.subscribe(TOPIC, qos=1)
def on_message(client, userdata, msg):
row = (
utc_now(),
msg.topic,
msg.payload.decode("utf-8", errors="replace"),
msg.qos,
int(msg.retain),
)
try:
inbox.put_nowait(row)
except queue.Full:
print("Queue full, dropping message") # la base de datos no da abasto
def on_disconnect(client, userdata, disconnect_flags, reason_code, properties):
print(f"Disconnected: {reason_code}, paho will reconnect")
def open_db(path: str) -> sqlite3.Connection:
db = sqlite3.connect(path)
db.execute("PRAGMA journal_mode=WAL") # los lectores no bloquean al escritor
db.execute("PRAGMA synchronous=NORMAL") # seguro con WAL y mucho más rápido
db.executescript(SCHEMA)
return db
def flush(db: sqlite3.Connection, batch: list) -> None:
if not batch:
return
with db: # una transacción por lote
db.executemany(
"INSERT INTO messages (received_at, topic, payload, qos, retained) VALUES (?, ?, ?, ?, ?)",
batch,
)
print(f"Stored {len(batch)} messages")
batch.clear()
def stop(*_):
global running
running = False
def main():
signal.signal(signal.SIGINT, stop)
signal.signal(signal.SIGTERM, stop)
db = open_db(DB_PATH)
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"sqlite-{uuid.uuid4().hex[:8]}")
# Broker privado de ConnectMQ:
# client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD")
# client.tls_set(); y usa el puerto 8883
client.on_connect = on_connect
client.on_message = on_message
client.on_disconnect = on_disconnect
client.reconnect_delay_set(min_delay=1, max_delay=60)
client.connect_async(BROKER, PORT, keepalive=60)
client.loop_start()
batch: list = []
last_flush = time.monotonic()
try:
while running:
try:
batch.append(inbox.get(timeout=0.2))
except queue.Empty:
pass
if len(batch) >= BATCH_SIZE or time.monotonic() - last_flush >= FLUSH_INTERVAL:
flush(db, batch)
last_flush = time.monotonic()
finally:
client.disconnect()
client.loop_stop()
while not inbox.empty():
batch.append(inbox.get_nowait())
flush(db, batch)
db.close()
print("Logger stopped cleanly")
if __name__ == "__main__":
main()
Ejecútalo con python mqtt_sqlite_logger.py. En otra terminal, publica algunos datos de prueba:
for i in $(seq 1 20); do
mosquitto_pub -h broker.connectmq.com -p 1883 -i sqlite-test-pub \
-t "connectmq/demo/k7f3q2/sensors/room$((i % 2 + 1))" \
-m "{\"temperature\": $((20 + RANDOM % 5)).$((RANDOM % 10)), \"humidity\": $((40 + RANDOM % 10))}"
sleep 1
done
El logger imprime "Stored 1 messages" más o menos una vez por segundo. Detenlo con Ctrl+C y la cola pendiente se vuelca antes de salir.
Cómo se mantiene fiable el logger
Marcas de tiempo. El logger registra cuándo se recibió el mensaje, en UTC y con milisegundos. Los relojes de los dispositivos derivan y algunos ni siquiera tienen reloj, así que la marca de recepción es una referencia fiable. Si tu payload incluye su propia marca de tiempo, guarda ambas. La guía de JSON con MQTT en Python muestra cómo añadirla en el lado del publicador.
Modo WAL. El write-ahead logging permite que un dashboard o la CLI sqlite3 lean la base de datos mientras el logger escribe. Sin él, las lecturas largas pueden bloquear las inserciones.
Cola acotada. Si el disco se atasca, la cola crece hasta 50.000 mensajes y después descarta los nuevos mostrando un aviso, en lugar de agotar la memoria. Ajusta el tamaño a tu tráfico.
Reconexiones. connect_async() junto con loop_start() sigue reintentando en segundo plano, y la suscripción se restaura en on_connect. Los mensajes publicados mientras el logger está desconectado se pierden salvo que uses una sesión persistente. Con MQTT 5 puedes fijar un session expiry interval y clean_start=False para que el broker encole los mensajes QoS 1 por ti; el broker público conserva las sesiones hasta 30 minutos. La guía de reconexión de paho-mqtt explica los detalles.
Retained messages. En cada reconexión el broker vuelve a enviar los retained messages, así que los verás guardados más de una vez. La columna retained te permite excluirlos del análisis.
Paso 3: consulta tu historial MQTT
Abre la base de datos con la CLI de SQLite (sqlite3 mqtt_messages.db) o con cualquier navegador SQLite y prueba estas consultas.
Los últimos 10 mensajes de un topic:
SELECT received_at, payload
FROM messages
WHERE topic = 'connectmq/demo/k7f3q2/sensors/room1'
ORDER BY received_at DESC
LIMIT 10;
Último valor por topic, usando el comportamiento documentado de SQLite para columnas sueltas con MAX():
SELECT topic, payload, MAX(received_at) AS last_seen
FROM messages
WHERE retained = 0
GROUP BY topic;
Medias por hora a partir de payloads JSON, con las funciones JSON integradas de SQLite:
SELECT strftime('%Y-%m-%d %H:00', received_at) AS hour,
topic,
ROUND(AVG(json_extract(payload, '$.temperature')), 2) AS avg_temperature,
COUNT(*) AS samples
FROM messages
WHERE topic LIKE 'connectmq/demo/k7f3q2/sensors/%'
AND json_valid(payload)
GROUP BY hour, topic
ORDER BY hour;
Conservar solo los últimos siete días y recuperar espacio:
DELETE FROM messages
WHERE received_at < strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-7 days');
VACUUM;
Como received_at se guarda en un formato ISO 8601 fijo, la simple comparación de cadenas ordena correctamente y el filtro por rango de tiempo puede usar el índice. Puedes lanzar la consulta de retención desde cron o desde el propio logger una vez al día.
Para trabajar con los datos en Python, pandas.read_sql_query("SELECT ...", sqlite3.connect("mqtt_messages.db")) devuelve un DataFrame. El tutorial de MQTT y pandas cubre el análisis y el remuestreo.
Lista de comprobación para resolver problemas
- "database is locked": otro proceso mantiene una transacción de escritura. Asegúrate de que el modo WAL está activado y de que solo un logger escribe en el archivo.
- No se guarda nada: confirma que el filtro de topic coincide con el topic del publicador; pruébalo con
mosquitto_sub -h broker.connectmq.com -p 1883 -i check-k7f3q2 -t 'connectmq/demo/k7f3q2/#' -v. json_extractdevuelve NULL: el nombre de la clave o la ruta JSON no coinciden, o el payload no es JSON. Revisa antes algunas filas en bruto.- La base de datos crece muy rápido: añade la tarea de retención y plantéate guardar solo los valores que cambian en el lado del publicador.
- El logger descarta mensajes: aumenta
BATCH_SIZE, mueve la base de datos a un almacenamiento más rápido o limita la suscripción a los topics que realmente necesitas.
Del prototipo a producción
En instalaciones reales, ejecuta el logger como servicio (systemd en Linux) para que se reinicie tras un reinicio del equipo, y apúntalo a un broker privado en lugar del público. Con ConnectMQ creas un broker en el portal, copias el usuario, la contraseña y el client ID generados y activas TLS en el puerto 8883; el tutorial de MQTT con TLS en Python muestra la configuración exacta. El broker de ConnectMQ no almacena por sí mismo el historial de mensajes, así que un suscriptor externo como este es el patrón honesto y recomendado. Si necesitas un punto de partida para el código del cliente, consulta el tutorial de paho mqtt en Python y, para otros lenguajes, el resumen de bibliotecas cliente MQTT. La documentación de sqlite3 de Python trata en profundidad las transacciones y las opciones de conexión.
Empieza con ConnectMQ
ConnectMQ es un broker MQTT gestionado con MQTT 3.1.1 y 5.0, TLS y WebSocket, además de un explorador de topics en vivo para comprobar qué envían tus dispositivos antes de registrarlo. Empieza hoy a registrar datos desde broker.connectmq.com y pasa después a tu propio broker MQTT en la nube. Crea una cuenta gratuita en el plan Serverless, sin tarjeta de crédito.
Preguntas frecuentes
¿Puede SQLite manejar datos MQTT continuos?
Sí, para un único logger a tasas moderadas. Agrupar las inserciones en transacciones y activar el modo WAL permite a SQLite absorber un flujo constante de muchos sensores en hardware modesto.
¿Por qué no insertar directamente dentro de on_message?
Hacer commit en el callback bloquea el bucle de red de paho con la E/S de disco. Bajo carga eso retrasa los pings de keepalive y los acuses de recibo, así que una cola con un escritor separado es más fiable.
¿Cómo evito perder mensajes mientras el logger se reinicia?
Usa QoS 1 con una sesión persistente de MQTT 5 (clean_start=False y un session expiry interval) para que el broker encole los mensajes mientras estás desconectado, dentro de su límite de expiración de sesión.
¿Puedo guardar payloads binarios en SQLite?
Sí. Cambia la columna payload a BLOB e inserta msg.payload directamente en lugar de decodificarlo a texto.
