Los datos de series temporales industriales son el flujo de mediciones con timestamp que máquinas, sensores y controladores producen cada segundo de cada turno. Gestionarlos bien implica decidir con qué frecuencia muestrear, qué descartar, cómo asignar timestamps y dónde guardarlos. Este artículo recorre cada una de esas decisiones y te ofrece código funcional para aplicar compresión deadband a los datos MQTT antes de que lleguen a un historian.
Qué diferencia a los datos de series temporales industriales
Todas las series temporales tienen la misma forma: un timestamp, un identificador y un valor. Los datos industriales añaden varias complicaciones que los datos de monitorización de IT rara vez tienen.
- Ritmos de cambio muy distintos. El nivel de un tanque cambia lentamente a lo largo de horas; una señal de vibración cambia miles de veces por segundo. Una única política de muestreo no sirve para ambas.
- Requisitos de retención largos. Las investigaciones de calidad, los registros regulatorios y el entrenamiento de modelos a menudo necesitan años de histórico, no semanas.
- La calidad importa tanto como el valor. Un valor procedente de un sensor desconectado no es una medición. Los sistemas industriales acompañan cada punto de un indicador de calidad.
- Datos de eventos y estados mezclados con datos analógicos. Estados de máquina, alarmas, IDs de lote y cambios de receta son escasos y deben combinarse con señales analógicas densas.
- Relojes que no controlas. PLC, gateways y servidores tienen cada uno su propio reloj, y derivan.
Acertar en esto desde el principio es mucho más barato que reparar años de histórico incoherente después.
Elegir las frecuencias de muestreo
La frecuencia de muestreo debe seguir la física de la señal y la pregunta que quieres responder, no el tiempo de ciclo por defecto del PLC.
| Tipo de señal | Comportamiento típico | Enfoque práctico |
|---|---|---|
| Temperaturas, niveles, presiones lentas | Cambian en segundos o minutos | Muestrear cada 1–10 s, comprimir con deadband |
| Caudales, corriente de motor, velocidades | Cambian en fracciones de segundo o segundos | Muestrear cada 100 ms–1 s, comprimir con deadband |
| Estado de máquina, alarmas, contadores | Eventos discretos | Publicar solo cuando cambian (informe por excepción) |
| Vibración, acústica, calidad de energía | Cambian en milisegundos | Muestrear a kHz en el edge, publicar características (RMS, pico, bandas) |
Dos reglas ayudan. Primera, el principio de Nyquist: para ver una variación, muestrea al menos al doble de su velocidad y, en la práctica, varias veces más rápido. Segunda, no transmitas por MQTT formas de onda brutas de alta frecuencia de forma continua. Calcula características en el edge y publica esas características; envía instantáneas en bruto solo bajo demanda o ante eventos.
Para los datos discretos, el informe por excepción es el patrón natural. El artículo sobre report by exception explica cómo reduce el tráfico sin perder información.
Compresión: conservar la señal y descartar el ruido
En los historians industriales, la compresión suele significar un filtrado con pérdida en la ingesta, no archivos zip. El objetivo es almacenar menos puntos sin dejar de poder reconstruir la señal dentro de una tolerancia conocida.
Filtrado por deadband (excepción)
El método más sencillo: guardar un punto nuevo solo cuando el valor difiere del último almacenado en más de un umbral. Un deadband de 0,2 degC en una temperatura significa que las fluctuaciones inferiores a 0,2 degC no se guardan. La mayoría de las implementaciones también guardan un punto tras un tiempo máximo aunque nada haya cambiado (un heartbeat), para que los consumidores distingan una señal estable de una muerta.
Compresión swinging door
El swinging door trending, utilizado en muchos historians comerciales, conserva un punto solo cuando una línea recta desde el último punto almacenado ya no puede representar los valores intermedios dentro de la tolerancia. Conserva las pendientes mucho mejor que un deadband simple, de modo que las rampas se guardan con pocos puntos en lugar de muchos.
Elegir las tolerancias
Fija la tolerancia en relación con la precisión del sensor, no con un porcentaje arbitrario. Filtrar por debajo del nivel de ruido del sensor almacena ruido; filtrar muy por encima pierde cambios reales del proceso. Documenta la tolerancia por tipo de tag en tu modelo de datos IIoT para que se revise como cualquier otro atributo del modelo.
Compresión deadband de datos MQTT en Python
Este suscriptor lee telemetría JSON desde MQTT, aplica un deadband con heartbeat y escribe los puntos que sobreviven en InfluxDB line protocol por la salida estándar, que puedes redirigir a un archivo o canalizar hacia un escritor. Usa el servidor MQTT público gratuito con un simulador de demo integrado; reserva ese servidor para pruebas y prototipos, nunca para datos de producción.
import json
import math
import random
import threading
import time
import uuid
import paho.mqtt.client as mqtt
BROKER = "broker.connectmq.com"
PREFIX = f"connectmq/demo/{uuid.uuid4().hex[:8]}/plant1/line1"
DEADBAND = {"temperature": 0.2, "pressure": 0.05}
HEARTBEAT_S = 60
last_stored = {} # topic -> (valor, ts_segundos)
def keep(topic, metric, value, ts):
prev = last_stored.get(topic)
if prev is None:
return True
band = DEADBAND.get(metric, 0.0)
return abs(value - prev[0]) > band or ts - prev[1] >= HEARTBEAT_S
def on_connect(client, userdata, flags, reason_code, properties):
client.subscribe(f"{PREFIX}/+/+", qos=1)
def on_message(client, userdata, msg):
asset, metric = msg.topic.split("/")[-2:]
p = json.loads(msg.payload)
ts = p["ts"] / 1000
if keep(msg.topic, metric, p["value"], ts):
last_stored[msg.topic] = (p["value"], ts)
print(f'{metric},asset={asset},quality={p["quality"]} value={p["value"]} {int(p["ts"] * 1e6)}')
def simulate(pub):
t0 = time.time()
while True:
t = time.time() - t0
temp = 60 + 5 * math.sin(t / 120) + random.gauss(0, 0.05)
pres = 2.5 + random.gauss(0, 0.01)
ts = int(time.time() * 1000)
for metric, val in [("temperature", temp), ("pressure", pres)]:
body = {"value": round(val, 3), "ts": ts, "quality": "GOOD"}
pub.publish(f"{PREFIX}/pump1/{metric}", json.dumps(body), qos=0)
time.sleep(1)
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="tsdb-" + uuid.uuid4().hex[:8])
client.on_connect = on_connect
client.on_message = on_message
client.connect(BROKER, 1883, keepalive=60)
threading.Thread(target=simulate, args=(client,), daemon=True).start()
client.loop_forever()
Ejecútalo unos minutos y cuenta las líneas: la presión, que solo muestra ruido por debajo de su deadband, genera aproximadamente un punto por minuto (el heartbeat), mientras que la temperatura genera puntos a medida que se mueve la onda senoidal. Ese es exactamente el comportamiento que quieres de un historian: denso donde el proceso se mueve y escaso donde no.
Timestamps y orden
Asigna siempre el timestamp en el origen, lo más cerca posible de la medición, en UTC y con milisegundos. La hora de llegada al broker no es la hora de medición, sobre todo cuando un gateway edge almacena datos en búfer durante un corte de red y los reenvía después. El patrón store and forward depende de los timestamps de origen para colocar los datos tardíos en su sitio.
Sincroniza los relojes con NTP o PTP en los gateways y, siempre que sea posible, en los controladores. Cuenta con llegadas fuera de orden y asegúrate de que tu almacenamiento acepta escrituras en el pasado. Algunas bases de datos de series temporales gestionan bien las escrituras tardías; otras las penalizan.
Opciones de almacenamiento para datos de series temporales industriales
| Opción | Puntos fuertes | Cuidado con |
|---|---|---|
| Historian de proceso comercial | Compresión, calidad, modelos de activos y retención larga integrados | Licencias, acceso propietario |
| Base de datos de series temporales (InfluxDB, TimescaleDB, QuestDB) | Interfaces abiertas, consultas SQL o Flux, buenas tasas de escritura | Tú diseñas la compresión, la retención y el submuestreo |
| Data lake (Parquet sobre almacenamiento de objetos) | Almacenamiento a largo plazo barato, ideal para entrenar ML | No sirve para consultas en tiempo real |
| Combinación por niveles | Datos calientes en la TSDB, datos fríos en el lake | Dos sistemas que operar |
Muchas plantas combinan un historian o una TSDB para los datos recientes con un lake para la analítica a largo plazo. El artículo sobre el historian MQTT explica cómo construir la parte del historian directamente a partir de los topics MQTT.
Llevar los datos del broker al almacenamiento
Un broker es un enrutador, no una base de datos: los mensajes lo atraviesan y los retained messages solo conservan el último valor por topic. Para guardar el histórico ejecutas un escritor que se suscribe e inserta. Telegraf es una elección habitual:
[[inputs.mqtt_consumer]]
servers = ["ssl://broker.connectmq.com:8883"]
topics = ["connectmq/plant1/#"]
client_id = "YOUR_CLIENT_ID"
username = "YOUR_USERNAME"
password = "YOUR_PASSWORD"
qos = 1
data_format = "json"
json_time_key = "ts"
json_time_format = "unix_ms"
tag_keys = ["quality"]
[[inputs.mqtt_consumer.topic_parsing]]
topic = "connectmq/+/+/+/+"
tags = "_/site/line/asset/_"
measurement = "_/_/_/_/measurement"
[[outputs.influxdb_v2]]
urls = ["http://localhost:8086"]
token = "YOUR_INFLUX_TOKEN"
organization = "plant"
bucket = "telemetry"
Con ConnectMQ como broker MQTT gestionado, Telegraf es simplemente otro cliente autenticado sobre TLS en el puerto 8883. Usa QoS 1 en la suscripción del escritor para que una desconexión breve no pierda datos en silencio, y mantén persistente la sesión del escritor si quieres que los mensajes se encolen mientras se reinicia.
Lista de comprobación para datos de series temporales industriales
- Clasifica cada señal: analógica lenta, analógica rápida, discreta o de alta frecuencia.
- Fija el muestreo y el deadband por clase, según la precisión del sensor.
- Añade un heartbeat para distinguir las señales estables de las muertas.
- Asigna el timestamp en el origen, en UTC y con milisegundos.
- Publica la calidad con cada valor.
- Calcula características en el edge para las señales de kHz.
- Elige el almacenamiento caliente y frío y define la retención y el submuestreo.
- Prueba las escrituras tardías y fuera de orden antes de la puesta en marcha.
Para la arquitectura que rodea todo esto, consulta la guía de IoT industrial con MQTT.
Empieza con ConnectMQ
Crea un broker Serverless gratuito en app.connectmq.com, sin tarjeta de crédito, y conecta después tu gateway y un escritor de Telegraf o Python con las credenciales generadas. El explorador de topics en vivo de ConnectMQ te ayuda a comprobar las tasas de publicación y los payloads antes de fijar una política de muestreo y compresión.
Preguntas frecuentes
¿Qué son los datos de series temporales industriales?
Son datos de mediciones y eventos con timestamp producidos por equipos industriales, como temperaturas, presiones, velocidades, estados y contadores. Normalmente se almacenan con un indicador de calidad y se conservan durante largos periodos para operaciones, calidad y analítica.
¿Cuál es un buen valor de deadband?
Empieza cerca de la precisión o del nivel de ruido del sensor para esa medición y comprueba después que los cambios importantes del proceso siguen viéndose en los datos almacenados. Documenta el valor por tipo de tag y revísalo cuando cambien los sensores.
¿Puede un broker MQTT almacenar histórico de series temporales?
No. Un broker enruta mensajes y conserva como mucho el último valor retenido por topic. Guarda el histórico con un suscriptor como Telegraf, un conector de historian o tu propio escritor.
¿Debo transmitir por MQTT los datos brutos de vibración?
Normalmente no de forma continua. Calcula en el edge características como RMS, pico y energía por bandas y publícalas, y envía instantáneas de la forma de onda en bruto solo bajo petición o cuando un evento las desencadene.
