Historian MQTT: captura de series temporales industriales

Un historian MQTT es un suscriptor que escucha los topics MQTT de tu planta y guarda cada valor como serie temporal, para que los operadores puedan ver tendencias, comparar e investigar lo que el namespace solo muestra en el momento. En esta guía compararás los principales enfoques de historian, montarás dos pipelines que funcionan, Telegraf hacia InfluxDB y Python hacia TimescaleDB, y conocerás las decisiones de diseño que determinan si tu historian seguirá siendo fiable dentro de un año.

Qué hace un historian MQTT

Un historian de proceso clásico sondea tags de SCADA o de los PLC y los comprime en un archivo propietario. Un historian MQTT invierte el flujo: los dispositivos y gateways publican en un broker y el historian es simplemente un suscriptor más. Eso tiene consecuencias útiles:

Sin carga adicional en la planta. El historian nunca sondea un PLC; lee lo que ya se publica.
Un único punto de integración. Añadir una máquina al namespace la añade automáticamente al historian si la suscripción cubre su rama.
Contexto a partir de los topics. Con topics ISA-95, sitio, área, línea y máquina se convierten en tags o columnas sin mapeo manual.

Lo que el historian debe resolver es todo lo que el broker no hace: almacenamiento a largo plazo, consultas eficientes por rango de tiempo, submuestreo y retención. Un broker conserva el último valor retenido y colas de sesión de corta duración, no meses de histórico. Este reparto de tareas se explica con más detalle en Unified Namespace vs data lake.

Comparativa de enfoques de historian

Enfoque	Puntos fuertes	Contrapartidas	Encaja bien en
InfluxDB con Telegraf	Motor de series temporales específico, ingesta solo con configuración, los tags encajan bien con los niveles ISA-95	Lenguaje de consulta y herramientas separados de los datos relacionales	Arranque rápido, muchas métricas, dashboards en Grafana
TimescaleDB (PostgreSQL)	SQL completo, joins con datos relacionales como órdenes y lotes, ecosistema maduro	Tienes que escribir o configurar el suscriptor de ingesta	Equipos con conocimientos de SQL, joins con datos del MES
Historian de proceso comercial	Funciones industriales como compresión, event frames, clientes pensados para OT, soporte del fabricante	Coste de licencias; la ingesta MQTT puede requerir un conector	Entornos regulados o con historians ya implantados
Archivos en data lake	El almacenamiento a largo plazo más barato, cualquier motor analítico	No está pensado para tendencias interactivas de operador	Archivo a largo plazo junto a un historian caliente

No hay una única respuesta correcta. Muchas plantas mantienen un historian caliente con semanas o meses de datos de alta resolución para operadores e ingenieros, y vuelcan todo en un lake para la analítica a largo plazo.

Datos de ejemplo para pruebas

Ambos pipelines esperan topics como <prefix>/<site>/<area>/<line>/<machine>/telemetry con payloads JSON. Si tienes el simulador de gateway de cómo construir un Unified Namespace, ejecútalo. Si no, este bucle publica datos simulados en el servidor MQTT público gratuito. El área pública es legible por cualquiera, así que usa tu propio prefijo aleatorio y solo valores simulados:

PREFIX="connectmq/demo/h6n3"
while true; do
  TS=$(date -u +%Y-%m-%dT%H:%M:%SZ)
  TEMP=$(awk -v r=$RANDOM 'BEGIN{printf "%.1f", 38 + r % 60 / 10}')
  mosquitto_pub -h broker.connectmq.com -p 1883 -i hist-sim-h6n3 \
    -t "$PREFIX/dallas/packaging/line2/filler/telemetry" \
    -m "{\"ts\":\"$TS\",\"temp_c\":$TEMP,\"speed_bpm\":118}"
  sleep 5
done

Pipeline 1: Telegraf e InfluxDB

El plugin MQTT consumer de Telegraf se suscribe al broker y escribe en InfluxDB sin código propio. Arranca InfluxDB 2 en Docker con un token de desarrollo:

docker run -d --name influxdb -p 8086:8086 \
  -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=plant \
  -e DOCKER_INFLUXDB_INIT_BUCKET=historian \
  -e DOCKER_INFLUXDB_INIT_ADMIN_TOKEN=dev-token-change-me \
  influxdb:2.7

Después crea telegraf.conf. El análisis del topic convierte cada nivel ISA-95 en un tag de InfluxDB:

[agent]
  interval = "10s"
  flush_interval = "10s"

[[inputs.mqtt_consumer]]
  servers = ["tcp://broker.connectmq.com:1883"]
  topics = ["connectmq/demo/h6n3/+/+/+/+/telemetry"]
  qos = 1
  client_id = "telegraf-historian-h6n3"
  name_override = "telemetry"
  data_format = "json"
  json_time_key = "ts"
  json_time_format = "2006-01-02T15:04:05Z07:00"

  [[inputs.mqtt_consumer.topic_parsing]]
    topic = "connectmq/demo/h6n3/+/+/+/+/telemetry"
    tags = "_/_/_/site/area/line/machine/_"

[[outputs.influxdb_v2]]
  urls = ["http://localhost:8086"]
  token = "dev-token-change-me"
  organization = "plant"
  bucket = "historian"

Ejecútalo con telegraf --config telegraf.conf. Abre la interfaz de InfluxDB en el puerto 8086, explora el bucket historian y verás una medición telemetry con los tags site, area, line y machine y campos numéricos como temp_c. El parser JSON conserva por defecto los campos numéricos; enumera explícitamente los campos de texto con json_string_fields si los necesitas. La documentación del MQTT consumer de Telegraf recoge todas las opciones.

Para un broker privado de ConnectMQ, cambia el servidor a ssl://YOUR_BROKER_HOST:8883, añade username = "YOUR_USERNAME" y password = "YOUR_PASSWORD" y asigna client_id a YOUR_CLIENT_ID, ya que los client IDs están vinculados a las credenciales emitidas en el portal. Elimina el prefijo de demo y sus niveles de análisis cuando tus topics empiecen por el nombre de tu empresa.

Pipeline 2: Python y TimescaleDB

Cuando quieres SQL y joins con órdenes de producción o registros de lotes, TimescaleDB es una elección natural. Arráncalo en Docker:

docker run -d --name timescaledb -p 5432:5432 -e POSTGRES_PASSWORD=historian timescale/timescaledb:latest-pg16

El suscriptor siguiente guarda los datos en formato largo, una fila por métrica, lo que se adapta bien a máquinas que publican campos distintos. Instala las dependencias con pip install "paho-mqtt>=2.0" "psycopg[binary]".

import json
import os
from datetime import datetime, timezone

import paho.mqtt.client as mqtt
import psycopg

DSN = os.getenv("PG_DSN", "postgresql://postgres:historian@localhost:5432/postgres")
PREFIX = os.getenv("UNS_PREFIX", "connectmq/demo/h6n3")

conn = psycopg.connect(DSN, autocommit=True)
conn.execute("""
    CREATE TABLE IF NOT EXISTS telemetry (
        time    TIMESTAMPTZ NOT NULL,
        site    TEXT, area TEXT, line TEXT, machine TEXT,
        metric  TEXT NOT NULL,
        value   DOUBLE PRECISION
    )""")
conn.execute("SELECT create_hypertable('telemetry', 'time', if_not_exists => TRUE)")

def on_connect(client, userdata, flags, reason_code, properties):
    print("historian connected:", reason_code)
    client.subscribe(f"{PREFIX}/+/+/+/+/telemetry", qos=1)

def on_message(client, userdata, msg):
    if msg.retain:
        return  # los valores retenidos ya se guardaron cuando se publicaron por primera vez
    site, area, line, machine, _ = msg.topic[len(PREFIX) + 1:].split("/")
    try:
        data = json.loads(msg.payload)
    except ValueError:
        return
    ts = data.pop("ts", None)
    when = datetime.fromisoformat(ts.replace("Z", "+00:00")) if ts else datetime.now(timezone.utc)
    rows = [(when, site, area, line, machine, k, float(v))
            for k, v in data.items() if isinstance(v, (int, float)) and not isinstance(v, bool)]
    with conn.cursor() as cur:
        cur.executemany(
            "INSERT INTO telemetry (time, site, area, line, machine, metric, value) "
            "VALUES (%s, %s, %s, %s, %s, %s, %s)", rows)

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="pg-historian-h6n3")
client.on_connect = on_connect
client.on_message = on_message
client.connect(os.getenv("MQTT_HOST", "broker.connectmq.com"), 1883, keepalive=60)
client.loop_forever()

Consulta medias de cinco minutos por máquina con SQL estándar más el time_bucket de Timescale:

SELECT time_bucket('5 minutes', time) AS bucket, line, machine, avg(value) AS avg_temp
FROM telemetry
WHERE metric = 'temp_c' AND time > now() - interval '1 day'
GROUP BY bucket, line, machine
ORDER BY bucket;

Decidir qué almacenar

Guardar para siempre cada valor publicado parece seguro, pero ralentiza las consultas y los costes crecen sin hacer ruido. Acuerda una política de almacenamiento con operaciones e ingeniería antes de poner en marcha el historian:

Valores de proceso como temperaturas, presiones y velocidades: guárdalos en bruto a la frecuencia publicada durante unas semanas y después conserva agregados como el mínimo, el máximo y la media por minuto.
Contadores como piezas buenas y rechazadas: guarda cada cambio, porque lo que importa para el OEE y el rendimiento son las diferencias entre lecturas.
Estados y alarmas: guarda cada transición con su timestamp. Su volumen es bajo y son esenciales para el análisis de paradas.
Eventos como la finalización de lotes: guárdalos todos, idealmente en una tabla con sus propios campos y no como métricas genéricas.
Diagnósticos como el tiempo de actividad del gateway o la intensidad de señal: aplica una retención más corta salvo que estés investigando un problema concreto.

Dejar esto por escrito para cada namespace en tu estándar de topics hace que la configuración del historian siga el diseño del namespace en lugar de crecer de forma improvisada.

Decisiones de diseño que importan

Timestamps de origen antes que hora de llegada. Guarda el timestamp del payload para que los datos en búfer o retrasados queden en el momento correcto. Conserva también la hora de llegada si necesitas medir la latencia.
Sesiones persistentes. Da al historian un client ID fijo y una sesión persistente (clean_session=False o una expiración de sesión MQTT 5) para que el broker encole los mensajes QoS 1 durante un reinicio breve.
Búfer en el edge para cortes largos. Las colas del broker cubren huecos breves. Para cortes de WAN de varias horas, almacena en el gateway; consulta store and forward con MQTT.
Gestión de los retained messages. Al suscribirse, el broker reproduce los valores retenidos. Omítelos o deduplica por tiempo y métrica para que los reinicios no creen filas duplicadas.
Bandas muertas e informe por excepción. Publicar solo los cambios significativos reduce drásticamente el almacenamiento sin perder información.
Retención y submuestreo. Conserva los datos en bruto durante un periodo definido y los agregados continuos o los intervalos submuestreados durante más tiempo.
Escalado horizontal. Con las suscripciones compartidas de MQTT 5, varios workers del historian pueden compartir $share/historian/... para que cada mensaje se escriba una sola vez. En ConnectMQ, las suscripciones compartidas están disponibles a partir del plan Serverless de pago.

ConnectMQ no incluye una base de datos ni un conector de historian integrados, y no los necesita para este patrón: cualquiera de los suscriptores anteriores se conecta como un cliente MQTT normal. Para profundizar en el modelado de los propios datos, lee datos de series temporales industriales, y para la arquitectura general consulta la guía de MQTT para IoT industrial.

Alimenta tu historian MQTT desde ConnectMQ

Un historian solo es tan completo como el flujo que lo alimenta. ConnectMQ es un broker MQTT gestionado con MQTT 3.1.1 y 5.0, QoS 1 y 2, sesiones persistentes, TLS y un portal que entiende ISA-95, de modo que tu historian con Telegraf o Python tiene una fuente fiable. Crea un broker en el plan gratuito Serverless en app.connectmq.com, sin tarjeta de crédito, o descubre más sobre cómo operar un broker MQTT en la nube para tu planta.

Preguntas frecuentes

¿Un broker MQTT es un historian?

No. Un broker enruta mensajes y conserva el último valor retenido por topic. Un historian se suscribe al broker y guarda la serie temporal completa para consultarla.

¿Debo elegir InfluxDB o TimescaleDB?

Elige InfluxDB con Telegraf para el montaje más rápido, solo con configuración, y dashboards en Grafana. Elige TimescaleDB cuando quieras SQL y joins con datos relacionales como órdenes y lotes.

¿Puede un historian comercial leer desde MQTT?

Muchos historians comerciales ofrecen conectores MQTT o Sparkplug. Comprueba si el conector admite tu formato de payload y tu estructura de topics, y si puede gestionar los births y alias de Sparkplug si los usas.

¿Cómo evito perder datos cuando el historian se reinicia?

Usa un client ID fijo con sesión persistente y QoS 1 para que el broker encole los mensajes durante un breve periodo, y almacena en búfer en el edge para cortes más largos.

Historian de datos MQTT: captura de series temporales industriales