Tutorial de Node-RED con InfluxDB: almacenar y consultar datos MQTT

Node-RED e InfluxDB juntos te ofrecen un pipeline de telemetría pequeño pero completo: llegan mensajes MQTT, Node-RED les da forma e InfluxDB los almacena como series temporales que puedes consultar y graficar. En este tutorial de Node-RED con InfluxDB ejecutarás InfluxDB 2 en Docker, escribirás lecturas MQTT con fields y tags adecuados y las consultarás con Flux para dibujar un gráfico. Cada paso usa un broker gratuito, así que puedes probar toda la cadena en menos de media hora.

Arquitectura: entra MQTT, salen series temporales

El pipeline tiene tres roles, y mantenerlos separados facilita escalar más adelante:

1. Los dispositivos publican lecturas JSON en topics MQTT como connectmq/demo/a7f3k2/room1/telemetry.
2. El broker enruta esos mensajes a cada suscriptor. No guarda histórico; ese es el trabajo de la base de datos. Si esta distinción es nueva para ti, aquí tienes en detalle qué hace un broker MQTT.
3. Node-RED se suscribe, valida y convierte cada mensaje en un punto de InfluxDB, y después lo escribe. Una segunda rama consulta InfluxDB para los gráficos.

Node-RED encaja bien cuando también quieres enrutamiento, alertas o un dashboard en el mismo lugar. Si solo necesitas ingesta a gran volumen, el consumidor MQTT de Telegraf es una alternativa, tratada en la guía para almacenar datos MQTT en InfluxDB.

Paso 1: ejecuta InfluxDB 2 con Docker

La imagen oficial puede inicializar una organización, un bucket y un token de administrador en el primer arranque, lo que ahorra varios clics:

docker run -d --name influxdb -p 8086:8086 \
  -v influxdb-data:/var/lib/influxdb2 \
  -e DOCKER_INFLUXDB_INIT_MODE=setup \
  -e DOCKER_INFLUXDB_INIT_USERNAME=admin \
  -e DOCKER_INFLUXDB_INIT_PASSWORD=change-me-please \
  -e DOCKER_INFLUXDB_INIT_ORG=demo-org \
  -e DOCKER_INFLUXDB_INIT_BUCKET=sensors \
  -e DOCKER_INFLUXDB_INIT_RETENTION=30d \
  -e DOCKER_INFLUXDB_INIT_ADMIN_TOKEN=my-super-secret-token \
  influxdb:2.7

Abre http://localhost:8086, inicia sesión y encontrarás el bucket sensors con 30 días de retención. En producción, crea un token de API aparte con acceso de escritura solo a ese bucket, en lugar de darle a Node-RED el token de administrador. Este tutorial está pensado para InfluxDB 2.x; InfluxDB 3 cambia la parte de consultas, así que revisa sus notas de compatibilidad antes de copiar los ejemplos de Flux.

Paso 2: publica datos de prueba en MQTT

Este script simula dos salas y publica en broker.connectmq.com, el servidor MQTT público gratuito. Cambia a7f3k2 por tu propia cadena aleatoria; los topics del servidor público son visibles para todo el mundo, así que úsalo solo para pruebas.

# pip install "paho-mqtt>=2.0"
import json
import random
import time
import uuid

import paho.mqtt.client as mqtt

PREFIX = "connectmq/demo/a7f3k2"
ROOMS = {"room1": 22.0, "room2": 25.0}

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                     client_id=f"influx-sim-{uuid.uuid4().hex[:8]}")
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_start()

try:
    while True:
        for room in ROOMS:
            ROOMS[room] += random.uniform(-0.2, 0.2)
            payload = {
                "temperature": round(ROOMS[room], 2),
                "humidity": round(random.uniform(38, 60), 1),
                "ts": int(time.time() * 1000),
            }
            client.publish(f"{PREFIX}/{room}/telemetry", json.dumps(payload), qos=1)
        time.sleep(5)
except KeyboardInterrupt:
    pass
finally:
    client.loop_stop()
    client.disconnect()

Incluir una marca de tiempo del dispositivo (ts) es una buena práctica: si los mensajes se retrasan, la base de datos sigue registrando cuándo se tomó la lectura, no cuándo llegó.

Paso 3: instala node-red-contrib-influxdb

En Node-RED, abre Manage palette → Install e instala node-red-contrib-influxdb. Obtienes tres nodos: influxdb out para escrituras individuales, influxdb batch para muchos puntos a la vez e influxdb in para consultas. Todos comparten un nodo de configuración influxdb.

Configura el nodo de configuración:

Paso 4: mapea los payloads MQTT a fields y tags

InfluxDB distingue entre fields (los valores medidos, no indexados) y tags (metadatos indexados por los que filtras y agrupas). Acertar con esto es la decisión de diseño más importante de toda la configuración de Node-RED con InfluxDB:

• Pon las lecturas numéricas, como temperatura y humedad, en fields.
• Pon los identificadores de baja cardinalidad, como sala, línea o ID de dispositivo, en tags.
• Nunca pongas valores sin límite (marcas de tiempo, IDs aleatorios, texto libre) en tags; cada combinación única de tags crea una nueva serie y el uso de memoria crece con ella.

Añade un nodo mqtt in suscrito a connectmq/demo/a7f3k2/+/telemetry, con salida "a parsed JSON object", y después este nodo function. El nodo influxdb out acepta un array de dos objetos: primero los fields y después los tags.

// Nodo function "to influx point"
const p = msg.payload;
if (typeof p !== "object" || p === null) {
    return null;
}

const fields = {};
if (typeof p.temperature === "number") fields.temperature = p.temperature;
if (typeof p.humidity === "number") fields.humidity = p.humidity;
if (Object.keys(fields).length === 0) {
    node.warn("No numeric fields in " + msg.topic);
    return null;
}

// connectmq/demo/a7f3k2/room1/telemetry -> room1
const room = msg.topic.split("/")[3];

// Usar la marca de tiempo del dispositivo si existe (milisegundos)
if (typeof p.ts === "number") {
    fields.time = new Date(p.ts);
}

msg.payload = [fields, { room: room, source: "node-red" }];
msg.measurement = "environment";
return msg;

Conéctalo a un nodo influxdb out con la organización demo-org, el bucket sensors y la precisión de tiempo en milisegundos. Deja vacío el measurement en el nodo para que se use msg.measurement. Haz deploy, espera un minuto y abre el Data Explorer en la interfaz de InfluxDB: aparecerá el measurement environment con los fields temperature y humidity y un tag room.

Si tu versión del nodo no acepta un field time dentro del objeto de fields, elimina esas dos líneas; InfluxDB usará entonces la hora de escritura.

Paso 5: consulta con Flux y grafica el resultado

Consultar es igual de sencillo. Un nodo inject se dispara cada minuto hacia un nodo function que construye la consulta Flux:

// Nodo function "build flux query"
msg.query = `
from(bucket: "sensors")
  |> range(start: -6h)
  |> filter(fn: (r) => r._measurement == "environment" and r._field == "temperature")
  |> aggregateWindow(every: 1m, fn: mean, createEmpty: false)
  |> keep(columns: ["_time", "_value", "room"])
`;
return msg;

Envíala a un nodo influxdb in con la organización demo-org. El nodo devuelve un array de filas, cada una con _time, _value y las columnas de tags. Conviértelas en puntos del gráfico, una serie por sala:

// Nodo function "rows to chart"
const rows = Array.isArray(msg.payload) ? msg.payload : [];
const byRoom = {};

for (const r of rows) {
    if (!byRoom[r.room]) byRoom[r.room] = [];
    byRoom[r.room].push({ x: Date.parse(r._time), y: Math.round(r._value * 100) / 100 });
}

const out = [{ payload: [] }];   // vaciar el gráfico antes de redibujar
for (const room of Object.keys(byRoom)) {
    out.push({ topic: room, payload: byRoom[room] });
}
return [out];

Conecta esto a un ui-chart de Dashboard 2.0 con las series desde msg.topic, la clave X x y la clave Y y. Como ahora los datos vienen de la base de datos, el gráfico sobrevive automáticamente a los reinicios. La guía de histórico de gráficos en Node-RED compara esto con opciones más ligeras si InfluxDB es más de lo que necesitas, y el tutorial de Dashboard 2.0 explica los ajustes del gráfico.

Lista de comprobación para producción

1. Usa un token de InfluxDB dedicado con acceso de escritura a un bucket para el flujo de escritura y acceso de lectura para el flujo de consulta.
2. Define una política de retención del bucket acorde con el tiempo que necesitas los datos en bruto; si necesitas tendencias largas, reduce la resolución de los datos antiguos en un segundo bucket con una task de InfluxDB.
3. Mantén los tags de baja cardinalidad y coherentes entre dispositivos.
4. Valida los payloads antes de escribir, como hace el nodo function, para que un dispositivo defectuoso no pueda llenar el bucket de cadenas.
5. Usa QoS 1 en la suscripción y una sesión persistente para que los reinicios breves de Node-RED no pierdan lecturas.
6. Usa el nodo influxdb batch si escribes cientos de puntos por segundo, agrupando antes los mensajes con un nodo join.
7. Monitoriza los errores de escritura: añade un nodo catch limitado al nodo influxdb out y envía los errores a un nodo debug o a una alerta.

Solución de problemas

El conflicto de tipo de field merece atención: InfluxDB fija el tipo de un field en la primera escritura dentro de un shard. Si un dispositivo envía 22 y más tarde 22.5, el desajuste entre entero y float se rechaza. Como JSON no los distingue, los sensores que envían números enteros de vez en cuando provocan este error. Enviar los valores de forma coherente desde el firmware (por ejemplo, siempre con un decimal) y comprobar el tipo del field en el Data Explorer tras las primeras escrituras evita sorpresas más adelante.

Protege también el lado MQTT

Una base de datos con tokens bien configurados no sirve de nada si cualquiera puede publicar lecturas falsas en los topics que ingiere. En el servidor público esa es exactamente la situación, y por eso es solo para pruebas. En producción, cambia el nodo de configuración mqtt-broker de Node-RED a un broker MQTT gestionado privado con TLS en el puerto 8883 y credenciales. En ConnectMQ, cada broker recibe un username, una password y un client ID generados, y los client IDs están vinculados a esas credenciales, así que un cliente cualquiera no puede suplantar tu flujo de ingesta.

Un broker MQTT en la nube también simplifica la red: los dispositivos en campo y Node-RED junto a InfluxDB se conectan hacia fuera al broker, así que no tienes que abrir puertos hacia la red de la base de datos. ConnectMQ no escribe en bases de datos por sí mismo; el patrón honesto es exactamente el de este artículo, un suscriptor Node-RED (o Telegraf) que escribe en InfluxDB. Para los fundamentos sobre brokers, comodines y nodos de configuración, consulta el tutorial de Node-RED con MQTT.

Empieza a almacenar tus datos MQTT con ConnectMQ

Crea una cuenta gratuita en app.connectmq.com con el plan Serverless, sin tarjeta de crédito. Crea un broker, copia las credenciales en tu nodo de configuración de Node-RED y tu pipeline de InfluxDB se alimentará de un flujo privado y cifrado con TLS que puedes inspeccionar en vivo en el explorador de topics del portal.

Preguntas frecuentes

¿Qué versiones de InfluxDB admite node-red-contrib-influxdb?

Admite InfluxDB 1.x con InfluxQL, 1.8 con Flux y 2.x con tokens, organizaciones y buckets. Elige la versión en el nodo de configuración para que los nodos muestren las opciones correctas.

¿Debo escribir cada mensaje MQTT en InfluxDB?

Escribe cada lectura que puedas necesitar más adelante, pero reduce la resolución de los datos de alta frecuencia en un bucket aparte para las tendencias a largo plazo. Si los dispositivos publican más rápido de lo que necesitas, promedia en Node-RED antes de escribir.

¿Cuál es la diferencia entre fields y tags en InfluxDB?

Los fields guardan los valores medidos y no están indexados. Los tags guardan metadatos como la sala o el ID del dispositivo, están indexados y deben tener un número limitado de valores distintos.

¿Puedo usar Grafana en lugar de un gráfico de Node-RED?

Sí. Apunta Grafana al mismo bucket de InfluxDB y usa la consulta Flux de esta guía. Node-RED sigue encargándose de la ingesta MQTT mientras Grafana se ocupa de la visualización.