Node-RED con MySQL: registrar datos MQTT en MySQL o MariaDB con consultas parametrizadas

Node-RED y MySQL forman una pareja práctica cuando quieres guardar datos de sensores MQTT en una base de datos que tu equipo ya sabe consultar, respaldar y usar para informes. En este tutorial crearás una tabla para las lecturas, te suscribirás a topics MQTT en Node-RED, insertarás cada mensaje con consultas parametrizadas, pasarás a inserciones por lotes cuando crezca la tasa de mensajes y configurarás la retención para que la tabla no crezca indefinidamente. Todo funciona igual en MariaDB.

¿Por qué registrar datos MQTT en MySQL?

Un broker MQTT mueve mensajes; no es un historiador. Los retained messages te dan el último valor de cada topic, pero no los valores de la semana pasada. Para crear informes, comparar turnos o alimentar una herramienta de BI, necesitas el histórico en algún lugar duradero.

Las bases de datos de series temporales como InfluxDB son una opción natural para la telemetría de alta frecuencia, y lo tratamos en Node-RED con InfluxDB. Aun así, MySQL y MariaDB ganan en muchas situaciones: la empresa ya los usa, las herramientas de informes se conectan a ellos de forma nativa, necesitas cruzar lecturas con datos relacionales como órdenes de trabajo o listas de activos, o la tasa de datos es moderada. Unos cientos de lecturas por minuto es terreno cómodo para una tabla bien indexada.

Paso 1: crea la base de datos y la tabla

Si todavía no tienes un servidor de base de datos, arranca MariaDB en Docker para las pruebas:

docker run -d --name iot-mariadb -p 3306:3306 \
  -e MARIADB_ROOT_PASSWORD=change-root \
  -e MARIADB_DATABASE=iot \
  -e MARIADB_USER=nodered \
  -e MARIADB_PASSWORD=change-me \
  mariadb:11

Después crea la tabla. Una fila por lectura, conservando el topic original para la trazabilidad y el payload completo para cualquier cosa que no hayas previsto:

CREATE TABLE mqtt_readings (
  id          BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
  received_at DATETIME(3)  NOT NULL,
  topic       VARCHAR(255) NOT NULL,
  device      VARCHAR(64)  NOT NULL,
  metric      VARCHAR(64)  NOT NULL,
  value       DOUBLE       NULL,
  payload     TEXT         NULL,
  INDEX idx_device_metric_time (device, metric, received_at),
  INDEX idx_time (received_at)
);

Guarda las marcas de tiempo en UTC y conviértelas en tus informes. Mezclar la hora local y los cambios de horario de verano en una tabla de series temporales es una fuente segura de huecos confusos y horas duplicadas.

Concede al usuario de Node-RED solo lo que necesita:

GRANT SELECT, INSERT, DELETE ON iot.mqtt_readings TO 'nodered'@'%';

Paso 2: instala el nodo MySQL de Node-RED

Instala node-red-node-mysql desde Menu → Manage palette. Añade un nodo mysql con una configuración de base de datos: host, puerto (3306), usuario, contraseña, nombre de la base de datos y zona horaria. Establece la zona horaria en Z o +00:00 para que las fechas se interpreten como UTC.

El nodo sigue un contrato sencillo:

• msg.topic contiene la sentencia SQL.
• msg.payload contiene los parámetros: un array para los marcadores ? o un objeto para los marcadores con nombre :name.
• El resultado vuelve en msg.payload: filas para un SELECT, o un objeto de resultado con affectedRows e insertId para un INSERT.

Paso 3: suscríbete a MQTT

Añade un nodo mqtt in conectado a tu broker. Para las pruebas, usa el servidor MQTT público gratuito en broker.connectmq.com, puerto 1883, un client ID único, y suscríbete a connectmq/demo/db91kz/+/+ con QoS 1 y la salida en detección automática. Los dos niveles comodín capturan el dispositivo y la métrica, por ejemplo connectmq/demo/db91kz/boiler1/temperature. Sustituye db91kz por tu propia cadena aleatoria; todo lo que hay en el servidor público es visible para otros, así que úsalo solo para pruebas.

Paso 4: inserta con consultas parametrizadas

Esta es la regla más importante del tutorial: nunca construyas SQL concatenando payloads MQTT en una cadena. Los topics y los payloads vienen de dispositivos y, en un broker compartido o mal configurado, potencialmente de cualquiera. Un payload como 1); DROP TABLE mqtt_readings; -- debe acabar como texto inofensivo en una columna, no ejecutado.

Añade un nodo function entre el nodo mqtt in y el nodo mysql:

// Topic: connectmq/demo/db91kz/<device>/<metric>
const mqttTopic = msg.topic;
const parts = mqttTopic.split("/");
if (parts.length !== 5) return null;

const device = parts[3].substring(0, 64);
const metric = parts[4].substring(0, 64);

// Aceptar números, cadenas numéricas u objetos {"value": ...}
let raw = msg.payload;
let value = null;
if (typeof raw === "number") {
    value = raw;
} else if (typeof raw === "string" && raw.trim() !== "" && !isNaN(Number(raw))) {
    value = Number(raw);
} else if (typeof raw === "object" && raw !== null && typeof raw.value === "number") {
    value = raw.value;
}

const payloadText = typeof raw === "string" ? raw : JSON.stringify(raw);
// Formato DATETIME(3) de MySQL en UTC: 2026-10-16 08:15:30.123
const receivedAt = new Date().toISOString().slice(0, 23).replace("T", " ");

msg.topic =
    "INSERT INTO mqtt_readings (received_at, topic, device, metric, value, payload) " +
    "VALUES (?, ?, ?, ?, ?, ?)";
msg.payload = [receivedAt, mqttTopic, device, metric, value, payloadText];
return msg;

Fíjate en que msg.topic se sobrescribe con la sentencia SQL; por eso el código guarda antes el topic MQTT en mqttTopic para la columna topic. Conecta la función al nodo mysql y añade después un nodo debug para ver el insertId de cada fila.

Probar la inserción

mosquitto_pub -h broker.connectmq.com -p 1883 -i sim-db91kz -t "connectmq/demo/db91kz/boiler1/temperature" -m "71.4"
mosquitto_pub -h broker.connectmq.com -p 1883 -i sim-db91kz -t "connectmq/demo/db91kz/boiler1/pressure" -m '{"value": 2.3, "unit": "bar"}'
mosquitto_pub -h broker.connectmq.com -p 1883 -i sim-db91kz -t "connectmq/demo/db91kz/boiler1/state" -m "running"

Después revisa la tabla:

SELECT received_at, device, metric, value, payload
FROM mqtt_readings
ORDER BY id DESC
LIMIT 10;

El tercer mensaje se guarda con value en NULL y running en la columna payload, que es exactamente lo que quieres para estados y otros datos no numéricos.

Paso 5: inserciones por lotes para tasas de mensajes más altas

Un INSERT por mensaje está bien para unos pocos mensajes por segundo. Por encima de eso, cada ida y vuelta y cada commit se acumulan. El procesamiento por lotes reduce la carga tanto en Node-RED como en la base de datos.

Cambia la función del paso 4 para que genere un array de fila en lugar de una sentencia SQL (establece msg.payload = [receivedAt, ...] y deja msg.topic como está) y añade después un nodo join en modo manual: combina cada msg.payload en un array y envía tras 100 mensajes o tras un timeout de 5 segundos. Comprueba con un nodo debug que el resultado es un array de filas (un array de arrays), no un único array plano. A continuación, añade esta función:

const rows = msg.payload;            // array de arrays de 6 elementos
if (!Array.isArray(rows) || rows.length === 0) return null;

const placeholders = rows.map(() => "(?, ?, ?, ?, ?, ?)").join(", ");
msg.topic =
    "INSERT INTO mqtt_readings (received_at, topic, device, metric, value, payload) VALUES " +
    placeholders;
msg.payload = rows.flat();
return msg;

Cada valor se sigue pasando como parámetro, así que el lote es tan seguro como la inserción individual. Si Node-RED se detiene mientras hay un lote pendiente, se pierden hasta cinco segundos de datos; para la mayoría de la monitorización es aceptable, y para cualquier caso más estricto, mantén las inserciones individuales.

Paso 6: retención y consultas útiles

Decide durante cuánto tiempo se conservan las lecturas en bruto. Una limpieza diaria sencilla en Node-RED usa un nodo inject configurado para repetirse a una hora fija, seguido de un nodo function:

msg.topic = "DELETE FROM mqtt_readings WHERE received_at < ? LIMIT 50000";
const cutoff = new Date(Date.now() - 90 * 24 * 3600 * 1000);
msg.payload = [cutoff.toISOString().slice(0, 23).replace("T", " ")];
return msg;

El LIMIT mantiene pequeño cada borrado para que no bloquee la tabla durante mucho tiempo; ejecútalo con más frecuencia si un día produce más filas que eso. El event scheduler de MySQL puede hacer el mismo trabajo dentro de la base de datos, si lo prefieres.

Dos consultas que usarás constantemente:

-- Promedio por hora y dispositivo de las últimas 24 horas
SELECT device,
       DATE_FORMAT(received_at, '%Y-%m-%d %H:00') AS hour,
       AVG(value) AS avg_value,
       MAX(value) AS max_value
FROM mqtt_readings
WHERE metric = 'temperature'
  AND received_at >= UTC_TIMESTAMP() - INTERVAL 24 HOUR
GROUP BY device, hour
ORDER BY device, hour;

-- Dispositivos que no han informado en 15 minutos
SELECT device, MAX(received_at) AS last_seen
FROM mqtt_readings
GROUP BY device
HAVING last_seen < UTC_TIMESTAMP() - INTERVAL 15 MINUTE;

La segunda consulta es una forma barata de encontrar dispositivos silenciosos. Para recibir un aviso en lugar de tener que mirar, combínala con el flujo de alerta por email en Node-RED. Para mostrar el histórico en un dashboard, la guía de histórico de gráficos en Node-RED muestra cómo consultarlo y graficarlo.

Solución de problemas en flujos de Node-RED con MySQL

Configuración para producción

Para datos reales, deja el servidor público. Crea un broker privado en ConnectMQ y configura el nodo del broker en Node-RED con el puerto 8883, TLS activado y los valores YOUR_USERNAME, YOUR_PASSWORD y YOUR_CLIENT_ID del portal. Usa una sesión persistente (clean session desactivado) con QoS 1 para que los reinicios breves de Node-RED no pierdan mensajes que el broker todavía puede entregar; en el plan gratuito Serverless el QoS llega hasta 1, que de todos modos es el nivel adecuado para el registro de datos.

Este reparto de responsabilidades es deliberado. ConnectMQ es un broker MQTT gestionado que entrega mensajes; no escribe en bases de datos por sí mismo. Node-RED es el cliente que habla con la base de datos, lo que mantiene tu esquema, tu retención y tu SQL totalmente bajo tu control. El tutorial de Node-RED con MQTT cubre el resto de ajustes del nodo del broker.

Registra tus datos MQTT con Node-RED y ConnectMQ

Crea un broker en el plan gratuito Serverless en app.connectmq.com, sin tarjeta de crédito, conecta Node-RED con TLS y empieza a construir un histórico que puedes consultar con SQL estándar. El explorador de topics en vivo del portal te ayuda a confirmar los topics y los formatos de payload antes de diseñar tu tabla en torno a ellos. Si todavía estás valorando opciones de alojamiento, nuestra guía sobre el broker MQTT en la nube compara los enfoques.

Preguntas frecuentes

¿Cómo paso parámetros al nodo MySQL de Node-RED?

Pon el SQL con marcadores ? en msg.topic y un array de valores en msg.payload. El nodo también acepta marcadores con nombre :name cuando msg.payload es un objeto.

¿Funciona con MariaDB?

Sí. node-red-node-mysql se conecta tanto a MySQL como a MariaDB, y la tabla, las inserciones y las consultas de este tutorial funcionan sin cambios en ambos.

¿Debo usar MySQL o InfluxDB para los datos MQTT?

Usa MySQL o MariaDB cuando necesites cruces con datos relacionales, herramientas de informes existentes o tasas de datos moderadas. Usa una base de datos de series temporales para telemetría de alta frecuencia y retención larga con reducción de resolución.

¿Puede ConnectMQ guardar mensajes directamente en una base de datos?

No. ConnectMQ es un broker MQTT sin integraciones con bases de datos. Ejecuta Node-RED u otro suscriptor que se conecte al broker y escriba en la base de datos, como se muestra en este tutorial.