El DataOps industrial es la disciplina de convertir los datos brutos de la tecnología operativa (OT) en información contextualizada y fiable y entregarla a cada consumidor que la necesita, de forma continua y no como un proyecto de integración puntual. En este artículo aprenderás las cuatro tareas que realiza una capa de DataOps, dónde se sitúa en una arquitectura basada en MQTT y cómo construir un pequeño servicio de contextualización que republica datos limpios en un namespace compartido.
Qué significa realmente el DataOps industrial
El término procede del DataOps de IT, que aplica ideas de DevOps (automatización, versionado, monitorización) a los pipelines de datos. En la planta el problema es distinto. Los datos los generan PLC, sensores y máquinas diseñados para el control, no para la analítica. Los tags no tienen unidades, los timestamps proceden de relojes distintos, máquinas idénticas usan nombres diferentes y nadie es responsable del mapeo entre ellos.
Sin una capa de DataOps, cada consumidor resuelve esos problemas por su cuenta. El equipo del historian escribe un mapeo, el integrador del MES escribe otro y el equipo de ciencia de datos limpia a mano una exportación CSV. Cada arreglo es local y frágil. El DataOps industrial traslada ese trabajo a una única capa gestionada entre las fuentes y los consumidores, de modo que la limpieza se hace una vez y se reutiliza en todas partes.
El resultado práctico es una capa de datos de planta en la que cada valor llega con su activo, unidad, timestamp, calidad y significado asociados, y en la que los cambios en esa capa se revisan y versionan como el código.
Las cuatro tareas de una capa de DataOps industrial
| Tarea | Qué hace | Ejemplo |
|---|---|---|
| Contextualización | Asocia identidad del activo, ubicación y significado a los valores brutos | PLC3.DB10.DBW4 pasa a ser berlin/line2/filler1/speed |
| Normalización | Convierte unidades, escalas, tipos de datos y nombres a un único estándar | Cuentas brutas 0–27648 escaladas a 0–200 bpm |
| Validación | Comprueba rangos, inactividad y calidad, y marca los datos erróneos | Un valor congelado durante 10 minutos se marca como STALE |
| Gobernanza | Es responsable del modelo, controla los cambios, documenta a los consumidores | Versiones del modelo, revisiones de cambios, reglas de acceso |
Contextualización
El contexto es lo que convierte un número en información. Una lectura de temperatura necesita saber a qué máquina pertenece, a qué parte de la máquina, qué producto se estaba fabricando y qué turno estaba de servicio. Parte del contexto es estático (activo, ubicación) y corresponde a la ruta del topic o al modelo del activo. Otra parte es dinámica (orden actual, receta, modo de operación) y lo mejor es publicarla como topics de estado separados que los consumidores puedan combinar con la telemetría. La guía de modelado de datos IIoT explica cómo estructurar ese contexto en tipos y jerarquías.
Normalización
La normalización hace que las cosas similares tengan el mismo aspecto. Dos motores de fabricantes distintos deberían informar ambos current en amperios, ambos state con la misma enumeración y ambos usar timestamps en milisegundos desde epoch. Los pasos de normalización típicos incluyen el escalado de valores brutos a unidades de ingeniería, la conversión de unidades, el renombrado, el mapeo de enumeraciones (códigos de estado del fabricante a un conjunto común) y la alineación de timestamps a UTC.
Validación
Los datos OT fallan de maneras concretas: los sensores se congelan, los valores saltan a los límites del rango analógico cuando se rompe un cable, los relojes derivan, los gateways reproducen datos en búfer fuera de orden. Una capa de DataOps debería detectar todo esto antes de que lo vean los consumidores. No hace falta borrar los datos sospechosos; a menudo es mejor conservarlos y asociarles un indicador de calidad, para que un historian pueda guardarlos mientras quien entrena un modelo los filtra.
Gobernanza
La gobernanza es la tarea menos técnica y la que más a menudo se omite. Alguien debe ser responsable de los tipos de activo, aprobar los cambios en las rutas de topics y saber qué aplicaciones dependen de qué datos. En la práctica eso significa mantener los modelos y los mapeos en control de versiones, revisar los cambios y publicar un registro de cambios que los consumidores puedan leer.
Dónde se sitúa el DataOps en una arquitectura MQTT
En una planta basada en MQTT, el broker es la columna vertebral de distribución y la lógica de DataOps vive en clientes a su alrededor. Una disposición habitual es esta:
PLCs / sensors
|
Edge gateway (protocol conversion, raw tags)
| publishes raw/...
MQTT broker <----> DataOps services (context, normalize, validate)
| republish to modeled namespace: site/area/line/cell/...
Consumers: historian, MES, dashboards, ML pipelines
El namespace bruto es útil para diagnosticar problemas, pero los consumidores solo deberían suscribirse al namespace modelado. Esta separación te permite corregir un mapeo sin tocar el gateway y añadir consumidores sin tocar el mapeo. Es el mismo principio que el del Unified Namespace: un único lugar para publicar y descubrir el estado actual, estructurado según la jerarquía de la planta.
Una nota de honestidad sobre las herramientas: el broker enruta mensajes; no ejecuta tu código de transformación. En un montaje con ConnectMQ, el servicio de contextualización es un cliente MQTT corriente (un proceso Python, un flujo de Node-RED o un producto comercial de DataOps) que se suscribe a los topics brutos y publica los modelados. ConnectMQ aporta el broker MQTT gestionado, la detección de topics ISA-95 y la decodificación y republicación de Sparkplug B en topics ISA-95, lo que cubre un paso de normalización habitual sin código propio.
Construir un pequeño servicio de contextualización
El siguiente servicio Python se suscribe a mensajes de tags brutos, aplica un archivo de mapeo, escala y valida los valores y los republica en un namespace ISA-95. Funciona contra el servidor MQTT público gratuito para experimentar; usa allí solo datos simulados, nunca datos de producción.
Primero, el mapeo, que viviría en control de versiones:
# mapping.yaml
modelVersion: 1.0.3
tags:
PLC3.DB10.DBW4:
path: berlin/line2/filler1/speed
unit: bpm
rawMin: 0
rawMax: 27648
engMin: 0
engMax: 200
validMin: 0
validMax: 180
PLC3.DB10.DBW6:
path: berlin/line2/filler1/temperature
unit: degC
rawMin: 0
rawMax: 27648
engMin: -20
engMax: 120
validMin: 5
validMax: 90
Después, el servicio:
import json
import sys
import time
import uuid
import paho.mqtt.client as mqtt
import yaml
BROKER = "broker.connectmq.com"
PREFIX = sys.argv[1] if len(sys.argv) > 1 else "connectmq/demo/" + uuid.uuid4().hex[:8]
STALE_SECONDS = 600
with open("mapping.yaml", encoding="utf-8") as f:
MAPPING = yaml.safe_load(f)
last_seen = {} # tag -> (valor bruto, primera vez que se vio este valor)
def scale(raw, m):
span = m["rawMax"] - m["rawMin"]
return m["engMin"] + (raw - m["rawMin"]) * (m["engMax"] - m["engMin"]) / span
def quality(tag, raw, value, m, now):
prev = last_seen.get(tag)
if prev is None or prev[0] != raw:
last_seen[tag] = (raw, now)
elif now - prev[1] > STALE_SECONDS:
return "STALE"
if not (m["validMin"] <= value <= m["validMax"]):
return "OUT_OF_RANGE"
return "GOOD"
def on_connect(client, userdata, flags, reason_code, properties):
client.subscribe(f"{PREFIX}/raw/#", qos=1)
print("listening on", f"{PREFIX}/raw/#")
def on_message(client, userdata, msg):
tag = msg.topic.split("/raw/", 1)[1]
m = MAPPING["tags"].get(tag)
if m is None:
client.publish(f"{PREFIX}/dataops/unmapped", json.dumps({"tag": tag}), qos=0)
return
raw = json.loads(msg.payload)
now = time.time()
value = round(scale(raw["v"], m), 2)
out = {
"value": value,
"unit": m["unit"],
"ts": raw.get("ts", int(now * 1000)),
"quality": quality(tag, raw["v"], value, m, now),
"modelVersion": MAPPING["modelVersion"],
}
client.publish(f"{PREFIX}/{m['path']}", json.dumps(out), qos=1)
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="dataops-" + uuid.uuid4().hex[:8])
client.on_connect = on_connect
client.on_message = on_message
client.connect(BROKER, 1883, keepalive=60)
client.loop_forever()
Instala las dependencias con pip install paho-mqtt pyyaml, arranca el servicio con el prefijo que prefieras y envíale un valor bruto:
python dataops.py connectmq/demo/plant42
mosquitto_pub -h broker.connectmq.com -p 1883 -i raw-test-42 \
-t "connectmq/demo/plant42/raw/PLC3.DB10.DBW4" -m '{"v": 16000, "ts": 1760000000000}'
mosquitto_sub -h broker.connectmq.com -p 1883 -i sub-test-42 -t "connectmq/demo/plant42/berlin/#" -v
El suscriptor ve una velocidad escalada de unos 115,7 bpm con unidad, timestamp, calidad y versión del modelo. Los tags sin mapear acaban en un topic dedicado en lugar de desaparecer en silencio; así descubres los tags nuevos que alguien añadió a un PLC.
Desplegar DataOps industrial: lista de comprobación
- Empieza por un consumidor que duela. Elige la integración que hoy requiere más limpieza manual.
- Separa los namespaces bruto y modelado desde el primer día e indica a los consumidores que usen solo el modelado.
- Pon los mapeos en control de versiones y despliégalos como código, con revisiones.
- Publica la calidad, no solo los valores. Deja que los consumidores decidan qué hacer con los datos inactivos o fuera de rango.
- Monitoriza la propia capa de DataOps. Controla los tags sin mapear, las tasas de mensajes y los recuentos de valores inactivos; publícalos también como topics.
- Haz explícita la propiedad de cada tipo de activo y de cada sitio.
- Documenta a los consumidores para que un cambio de modelo pueda contrastarse con quién depende de él.
Errores comunes
- Tratar el DataOps como una migración puntual. Las plantas cambian constantemente. Si nadie mantiene el mapeo, se degrada en pocos meses.
- Limpiar los datos dentro de cada dashboard. La lógica oculta en las herramientas de visualización no puede reutilizarse ni probarse.
- Descartar los datos erróneos. Borrar valores destruye pruebas. Márcalos en su lugar.
- Ignorar el tiempo. Sin timestamps de origen ni sincronización de relojes, combinar señales de distintas máquinas es adivinar. Datos de series temporales industriales explica por qué esto importa para el almacenamiento y la analítica.
Escalar más allá de una planta
Cuando varios sitios, por ejemplo plantas en Querétaro, Santiago y Valencia, publican en la misma estructura, la capa de DataOps se convierte en el contrato entre ellos. Un broker MQTT en la nube permite que los servicios de cada sitio publiquen los datos modelados en un único lugar, mientras que las cuotas por plan y el aislamiento multitenant mantienen separados los entornos. Para la visión arquitectónica general, la guía de IoT industrial con MQTT muestra cómo encajan gateways, brokers y consumidores, y el artículo sobre el historian MQTT cubre la parte del almacenamiento. Si además quieres visualizar la planta como gemelo digital 3D, el namespace modelado es la fuente natural de datos para la escena.
Ejecuta tu capa de DataOps en ConnectMQ
Crea un broker Serverless gratuito en app.connectmq.com, sin tarjeta de crédito. Apunta tu gateway edge al namespace bruto, ejecuta tu servicio de contextualización como cliente con YOUR_USERNAME, YOUR_PASSWORD y YOUR_CLIENT_ID, y usa el explorador de topics en vivo y la vista de dispositivos ISA-95 de ConnectMQ para confirmar que el namespace modelado tiene el aspecto que dicta tu modelo.
Preguntas frecuentes
¿Qué es el DataOps industrial?
El DataOps industrial es la práctica de contextualizar, normalizar, validar y gobernar los datos OT en una única capa gestionada, de modo que historians, MES, dashboards y analítica reciban la misma información limpia y bien descrita.
¿El DataOps industrial es un producto o una práctica?
Es una práctica que pueden apoyar ciertos productos. Puedes implementarla con plataformas comerciales de DataOps, flujos de Node-RED o servicios propios, siempre que los mapeos estén versionados, tengan responsable y se apliquen antes de que los datos lleguen a los consumidores.
¿El broker MQTT realiza las transformaciones?
No. El broker distribuye mensajes; la transformación se ejecuta en clientes MQTT que se suscriben a los topics brutos y publican los modelados. Además, ConnectMQ decodifica Sparkplug B y puede republicarlo en topics ISA-95.
¿Qué relación tiene el DataOps industrial con un Unified Namespace?
El Unified Namespace es donde se publican y descubren los datos modelados; el DataOps es el trabajo que hace que esos datos sean coherentes y fiables. Un namespace sin DataOps tiende a convertirse en una colección bien organizada de datos desordenados.
