MQTT a MongoDB: guardar mensajes de dispositivos con Node.js y colecciones de series temporales

Guardar MQTT en MongoDB encaja de forma natural cuando tus dispositivos envían JSON: los documentos entran tal como llegan, y las colecciones de series temporales mantienen el almacenamiento compacto y las consultas rápidas. En este tutorial construirás un suscriptor en Node.js con MQTT.js que agrupa en lotes los mensajes de los dispositivos procedentes de un broker en la nube y los escribe en una colección de series temporales de MongoDB, con expiración automática y consultas de agregación listas para usar.

Cómo encaja el pipeline

devices --MQTT--> ConnectMQ broker --MQTT--> Node.js subscriber --insertMany--> MongoDB time-series collection

La función del broker es enrutar mensajes; no escribe en bases de datos. El suscriptor es un cliente MQTT normal que ejecutas junto a tu base de datos, lo que mantiene el diseño sencillo y portable. En el lado izquierdo sirve cualquier broker MQTT gestionado; este tutorial usa ConnectMQ.

Por qué colecciones de series temporales

Una colección normal guarda cada lectura como un documento propio con su propia sobrecarga. Una colección de series temporales agrupa internamente las lecturas por un campo de metadatos y una ventana de tiempo, de modo que los datos IoT ocupan mucho menos disco y las consultas por rangos de tiempo recorren menos datos. También obtienes expireAfterSeconds para borrar automáticamente los datos antiguos. Las contrapartidas: los documentos son en la práctica de solo inserción (las actualizaciones y los borrados están limitados) y no se admiten índices únicos, así que, si necesitas deduplicar, debes hacerlo antes de insertar.

Paso 1: arrancar MongoDB

docker run -d --name mongo -p 27017:27017 \
  -v mongo-data:/data/db \
  mongo:7

Paso 2: crear la colección de series temporales

Los mensajes de los dispositivos llegan a <prefix>/<site>/<deviceId>/telemetry con un payload como este:

{"ts": "2026-10-16T09:30:00.000Z", "temperature": 23.4, "humidity": 41, "fw": "1.4.2"}

La colección guarda cada mensaje como un documento: un campo de tiempo ts, un objeto meta con los valores que identifican la serie (planta, dispositivo) y las mediciones. Créala una vez con un script de configuración.

setup.mjs:

import { MongoClient } from "mongodb";

const uri = process.env.MONGO_URI ?? "mongodb://localhost:27017";
const client = new MongoClient(uri);
await client.connect();
const db = client.db("iot");

try {
  await db.createCollection("telemetry", {
    timeseries: { timeField: "ts", metaField: "meta", granularity: "seconds" },
    expireAfterSeconds: 60 * 60 * 24 * 90, // conservar 90 días
  });
  console.log("created time-series collection iot.telemetry");
} catch (err) {
  if (err.codeName === "NamespaceExists") console.log("collection already exists");
  else throw err;
}

await db.collection("telemetry").createIndex({ "meta.deviceId": 1, ts: -1 });
await client.close();

npm init -y
npm install mqtt mongodb
node setup.mjs

Elige granularity según la frecuencia con la que informa cada dispositivo: seconds para intervalos de menos de un minuto, minutes para lecturas cada pocos minutos y hours para datos dispersos. Un buen metaField contiene valores que rara vez cambian dentro de una serie; nunca pongas las propias mediciones en meta.

Paso 3: el suscriptor de MQTT a MongoDB

El suscriptor mantiene ligero el callback MQTT: valida el mensaje, construye un documento y lo añade a un búfer en memoria. Un temporizador vacía el búfer con insertMany cada segundo, o de inmediato en cuanto hay 500 documentos esperando.

subscriber.mjs:

import mqtt from "mqtt";
import { MongoClient } from "mongodb";
import { randomBytes } from "node:crypto";

const MQTT_URL = process.env.MQTT_URL ?? "mqtts://broker.connectmq.com:8883";
const PREFIX = process.env.MQTT_TOPIC_PREFIX ?? "connectmq/demo/mongo-9d42a7";
const CLIENT_ID = process.env.MQTT_CLIENT_ID ?? `mongo-sink-${randomBytes(4).toString("hex")}`;
const MONGO_URI = process.env.MONGO_URI ?? "mongodb://localhost:27017";

const BATCH_SIZE = 500;
const FLUSH_MS = 1000;
const MAX_BUFFER = 50_000;

const mongo = new MongoClient(MONGO_URI);
await mongo.connect();
const telemetry = mongo.db("iot").collection("telemetry");

let buffer = [];
let flushing = false;

function toDocument(topic, payload) {
  // <prefix>/<site>/<deviceId>/telemetry
  const parts = topic.split("/");
  if (parts.length < 3 || parts.at(-1) !== "telemetry") return null;
  const deviceId = parts.at(-2);
  const site = parts.at(-3);

  let data;
  try {
    data = JSON.parse(payload.toString());
  } catch {
    return null;
  }
  if (typeof data !== "object" || data === null || Array.isArray(data)) return null;

  const { ts, ...fields } = data;
  const when = ts ? new Date(ts) : new Date();
  if (Number.isNaN(when.getTime())) return null;

  const doc = { ts: when, meta: { site, deviceId }, receivedAt: new Date() };
  for (const [key, value] of Object.entries(fields)) {
    if (key.startsWith("$") || key.includes(".")) continue; // nombres de campo inseguros
    if (["number", "string", "boolean"].includes(typeof value)) doc[key] = value;
  }
  return doc;
}

async function flush() {
  if (flushing || buffer.length === 0) return;
  flushing = true;
  const batch = buffer;
  buffer = [];
  try {
    await telemetry.insertMany(batch, { ordered: false });
    console.log(`inserted ${batch.length} documents`);
  } catch (err) {
    console.error("insert failed, re-queueing batch:", err.message);
    buffer = batch.concat(buffer).slice(-MAX_BUFFER);
  } finally {
    flushing = false;
  }
}

const timer = setInterval(flush, FLUSH_MS);

const client = mqtt.connect(MQTT_URL, {
  clientId: CLIENT_ID,
  username: process.env.MQTT_USERNAME,
  password: process.env.MQTT_PASSWORD,
  protocolVersion: 5,
  clean: false,                                   // sesión persistente
  properties: { sessionExpiryInterval: 1800 },    // el broker la conserva 30 min
  keepalive: 30,
  reconnectPeriod: 2000,
});

client.on("connect", (connack) => {
  console.log(`connected as ${CLIENT_ID}, session present: ${connack.sessionPresent}`);
  client.subscribe(`${PREFIX}/+/+/telemetry`, { qos: 1 }, (err) => {
    if (err) console.error("subscribe failed:", err.message);
  });
});

client.on("message", (topic, payload) => {
  const doc = toDocument(topic, payload);
  if (!doc) return;
  if (buffer.length >= MAX_BUFFER) {
    console.error("buffer full, dropping message");
    return;
  }
  buffer.push(doc);
  if (buffer.length >= BATCH_SIZE) flush();
});

client.on("error", (err) => console.error("mqtt error:", err.message));
client.on("reconnect", () => console.log("reconnecting..."));

async function shutdown() {
  console.log("shutting down");
  clearInterval(timer);
  await client.endAsync();
  await flush();
  await mongo.close();
  process.exit(0);
}
process.on("SIGINT", shutdown);
process.on("SIGTERM", shutdown);

Notas de diseño

• Lotes con insertMany y ordered: false. Un viaje de ida y vuelta por lote en lugar de por mensaje, y un único documento defectuoso no detiene el resto del lote.
• Un búfer acotado. Si MongoDB no está accesible, los lotes fallidos se vuelven a encolar, pero el búfer tiene un límite para que la memoria no pueda crecer indefinidamente.
• Una sesión persistente. clean: false con una expiración de sesión y un client ID fijo indica al broker que encole los mensajes QoS 1 mientras el suscriptor se reinicia. El servidor MQTT público gratuito conserva las sesiones hasta 30 minutos. Usa un MQTT_CLIENT_ID estable o tendrás una sesión nueva y vacía en cada arranque. Más información en sesiones persistentes en MQTT.
• Sanitización de campos. Se omiten las claves que empiezan por $ o contienen puntos, y solo se guardan valores primitivos, de modo que un dispositivo malicioso o defectuoso no puede inyectar estructuras extrañas en tus documentos.
• Marcas de tiempo del dispositivo. Usar el ts del dispositivo sitúa en el momento correcto las lecturas almacenadas mientras estaba desconectado; receivedAt registra cuándo las recibió el servidor, lo que ayuda a depurar problemas de reloj.

Dar forma a los documentos: ¿uno por mensaje o uno por métrica?

El suscriptor guarda un documento por mensaje MQTT, con cada medición como campo de primer nivel. Esa forma coincide con cómo publican los dispositivos, mantiene juntos los valores relacionados y hace triviales consultas como "temperatura y humedad en el mismo instante". Funciona bien cuando cada tipo de dispositivo envía un conjunto estable de campos.

La alternativa es un documento por métrica: { ts, meta: { deviceId, metric }, value }. Encaja en flotas donde los dispositivos informan de mediciones muy distintas, porque todos los documentos tienen la misma estructura y puedes indexar y agregar sobre meta.metric de forma genérica. El coste es tener más documentos y la necesidad de reagrupar los valores que llegaron juntos.

Elige una forma por colección y mantenla. Mezclar formas en una misma colección de series temporales complica las agregaciones y reduce las ventajas de almacenamiento. Si no lo tienes claro, empieza con un documento por mensaje; es más fácil separar después que fusionar.

Paso 4: probar con dispositivos simulados

Ejecuta el suscriptor y publica lecturas de prueba en el servidor público. El área pública es visible para todo el mundo, así que limítate a datos sintéticos:

node subscriber.mjs

# en otra terminal
for i in $(seq 1 30); do
  mosquitto_pub -h broker.connectmq.com -p 1883 -i mongo-sim-01 -q 1 \
    -t "connectmq/demo/mongo-9d42a7/plant1/sensor-01/telemetry" \
    -m "{\"temperature\": 2$((RANDOM % 10)).$((RANDOM % 10)), \"humidity\": 4$((RANDOM % 10))}"
  sleep 1
done

Después revisa los datos en mongosh:

use iot
db.telemetry.find().sort({ ts: -1 }).limit(3)

Paso 5: consultar los datos

Medias por minuto de un dispositivo en las últimas 24 horas:

db.telemetry.aggregate([
  { $match: { "meta.deviceId": "sensor-01", ts: { $gte: new Date(Date.now() - 864e5) } } },
  { $group: {
      _id: { $dateTrunc: { date: "$ts", unit: "minute" } },
      avgTemp: { $avg: "$temperature" },
      maxTemp: { $max: "$temperature" },
      samples: { $sum: 1 }
  } },
  { $sort: { _id: 1 } }
])

Última lectura por dispositivo, útil para una página de resumen de la flota:

db.telemetry.aggregate([
  { $sort: { "meta.deviceId": 1, ts: -1 } },
  { $group: { _id: "$meta.deviceId", last: { $first: "$$ROOT" } } }
])

Para dashboards a largo plazo, programa un job que escriba resúmenes por hora en una colección aparte con $merge, y deja que la colección en bruto expire a los 90 días.

Lista de comprobación para producción

1. Crea un broker privado en el portal de ConnectMQ y define MQTT_USERNAME, MQTT_PASSWORD y MQTT_CLIENT_ID con los valores generados; mantén mqtts:// en el puerto 8883.
2. Da al suscriptor sus propias credenciales. ConnectMQ vincula los client ID a las credenciales emitidas, y dos procesos con el mismo ID se desconectan mutuamente.
3. Ejecuta el suscriptor bajo un gestor de procesos (systemd, PM2, Docker con política de reinicio).
4. Usa un usuario de MongoDB con permisos de inserción solo en iot.telemetry.
5. Asume que QoS 1 puede entregar duplicados; si importan, deduplica en el suscriptor (por ejemplo, con una pequeña caché en memoria de IDs de mensaje), porque las colecciones de series temporales no pueden imponer unicidad.
6. Para repartir la carga entre varios suscriptores, usa suscripciones compartidas de MQTT 5 ($share/mongo/<prefix>/+/+/telemetry), disponibles en los planes Serverless de pago y superiores.
7. Vigila la tasa de inserción y el tamaño del búfer en los logs o las métricas.

Solución de problemas

Para una alternativa SQL con agregados continuos, consulta MQTT a TimescaleDB; para necesidades más ligeras, puede bastar con guardar datos MQTT en SQLite con Python o con MQTT a Google Sheets. La guía de bibliotecas cliente MQTT trata MQTT.js junto con otros clientes.

Empieza con ConnectMQ

Conecta tu pipeline de MongoDB a un broker privado en pocos minutos. Regístrate en app.connectmq.com para obtener un broker Serverless gratuito, sin tarjeta de crédito, exporta las credenciales generadas como variables de entorno y usa el explorador de topics en vivo del portal para revisar los payloads antes de que lleguen a tu base de datos.

Preguntas frecuentes

¿Puede ConnectMQ guardar mensajes MQTT en MongoDB automáticamente?

No, ConnectMQ es el broker MQTT y no tiene integraciones con bases de datos incorporadas. Ejecuta un suscriptor, como el servicio Node.js de este tutorial o un flujo de Node-RED, que se conecte como cliente MQTT y escriba en MongoDB.

¿Debo usar una colección de series temporales o una colección normal para los datos MQTT?

Usa una colección de series temporales para las lecturas de sensores de solo inserción: las almacena de forma compacta y admite expiración automática. Usa una colección normal para los datos que actualizas en el mismo documento, como la configuración de los dispositivos o el último estado de cada uno.

¿Cómo evito perder mensajes MQTT cuando MongoDB está caído?

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

¿Para qué sirve el metaField en una colección de series temporales de MongoDB?

Identifica cada serie, por ejemplo la planta y el ID del dispositivo. MongoDB agrupa internamente las lecturas con los mismos metadatos, y por eso las mediciones nunca deben ir en el metaField.