Industrial DataOps: Contextualizing and Governing Plant Data

Industrial DataOps is the discipline of turning raw operational technology (OT) data into contextualized, trustworthy information and delivering it to every consumer that needs it, continuously rather than as a one-off integration project. In this article you will learn the four jobs a DataOps layer performs, where it sits in an MQTT-based architecture, and how to build a small contextualization service that republishes clean data into a shared namespace.

What Industrial DataOps Actually Means

The term borrows from IT DataOps, which applies DevOps ideas (automation, versioning, monitoring) to data pipelines. In the plant, the problem is different. The data is generated by PLCs, sensors and machines that were designed for control, not for analytics. Tags lack units, timestamps come from different clocks, identical machines use different names, and nobody owns the mapping between them.

Without a DataOps layer, every consumer solves those problems on its own. The historian team writes one mapping, the MES integrator writes another, and the data science team cleans a CSV export by hand. Each fix is local and fragile. Industrial DataOps moves that work into one managed layer between the sources and the consumers, so the cleaning is done once and reused everywhere.

The practical result is a plant data layer where a value arrives with its asset, unit, timestamp, quality and meaning attached, and where changes to that layer are reviewed and versioned like code.

The Four Jobs of an Industrial DataOps Layer

Contextualization

Context is what turns a number into information. A temperature reading needs to know which machine it belongs to, which part of the machine, which product was running and which shift was on duty. Some context is static (asset, location) and belongs in the topic path or the asset model. Some is dynamic (current order, recipe, operating mode) and is best published as separate state topics that consumers can join with telemetry. The IIoT data modeling guide covers how to structure that context into types and hierarchies.

Normalization

Normalization makes similar things look the same. Two motors from different vendors should both report current in amperes, both report state with the same enumeration, and both use millisecond epoch timestamps. Typical normalization steps include raw-to-engineering scaling, unit conversion, renaming, enum mapping (vendor state codes to a common set) and timestamp alignment to UTC.

Validation

OT data fails in specific ways: sensors freeze, values jump to the limits of the analog range when a wire breaks, clocks drift, gateways replay buffered data out of order. A DataOps layer should catch these before consumers see them. You do not need to delete suspicious data; it is often better to keep it and attach a quality flag, so a historian can store it while a model trainer can filter it out.

Governance

Governance is the least technical job and the one most often skipped. Someone must own the asset types, approve changes to topic paths, and know which applications depend on which data. In practice that means keeping models and mappings in version control, reviewing changes, and publishing a changelog that consumers can read.

Where DataOps Sits in an MQTT Architecture

In an MQTT-based plant, the broker is the distribution backbone and the DataOps logic lives in clients around it. A common layout looks like this:

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

The raw namespace is useful for troubleshooting, but consumers should only subscribe to the modeled namespace. This separation lets you fix a mapping without touching the gateway, and lets you add consumers without touching the mapping. It is the same principle as the unified namespace: one place to publish and discover current state, structured by the plant hierarchy.

A note on honesty about tooling: the broker routes messages; it does not run your transformation code. In a ConnectMQ setup, the contextualization service is an ordinary MQTT client (a Python process, a Node-RED flow or a commercial DataOps product) that subscribes to raw topics and publishes modeled ones. ConnectMQ provides the managed MQTT broker, ISA-95 topic detection and Sparkplug B decoding and republishing into ISA-95 topics, which covers one common normalization step without custom code.

Building a Small Contextualization Service

The following Python service subscribes to raw tag messages, applies a mapping file, scales and validates the values, and republishes them into an ISA-95 namespace. It runs against the public MQTT test broker for experimentation; use only simulated data there.

First, the mapping, which would live in version control:

# 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

Then the service:

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 -> (raw value, first time this value was seen)


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()

Install the dependencies with pip install paho-mqtt pyyaml, start the service with a prefix of your choice, and feed it a raw value:

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

The subscriber sees a scaled speed of about 115.7 bpm with unit, timestamp, quality and model version. Unmapped tags land on a dedicated topic instead of disappearing silently, which is how you discover new tags that someone added to a PLC.

Rolling Out Industrial DataOps: A Checklist

1. Start with one consumer that hurts. Pick the integration that currently needs the most manual cleanup.
2. Separate raw and modeled namespaces from day one, and tell consumers to use only the modeled one.
3. Put mappings in version control and deploy them like code, with reviews.
4. Publish quality, not just values. Let consumers decide what to do with stale or out-of-range data.
5. Monitor the DataOps layer itself. Track unmapped tags, message rates and stale counts; publish them as topics too.
6. Make ownership explicit for every asset type and every site.
7. Document consumers so a model change can be checked against who depends on it.

Common Mistakes

• Treating DataOps as a one-time migration. Plants change constantly. If nobody maintains the mapping, it decays within months.
• Cleaning data inside each dashboard. Logic hidden in visualization tools cannot be reused or tested.
• Dropping bad data. Deleting values destroys evidence. Flag them instead.
• Ignoring time. Without source timestamps and clock synchronization, joining signals across machines is guesswork. Industrial time series data explains why this matters for storage and analytics.

Scaling Beyond One Plant

When several sites publish into the same structure, the DataOps layer becomes the contract between them. A cloud MQTT broker lets each site's services publish modeled data to one place, while per-plan quotas and multi-tenant isolation keep environments separate. For the bigger architectural picture, the industrial IoT with MQTT guide shows how gateways, brokers and consumers fit together, and the MQTT historian article covers the storage side.

Run Your DataOps Layer on ConnectMQ

Create a free Serverless broker at app.connectmq.com, no credit card required. Point your edge gateway at the raw namespace, run your contextualization service as a client with YOUR_USERNAME, YOUR_PASSWORD and YOUR_CLIENT_ID, and use the ConnectMQ live topic explorer and ISA-95 devices view to confirm that the modeled namespace looks the way your model says it should.

Frequently Asked Questions

What is industrial DataOps?

Industrial DataOps is the practice of contextualizing, normalizing, validating and governing OT data in one managed layer, so that historians, MES, dashboards and analytics all receive the same clean, well-described information.

Is industrial DataOps a product or a practice?

It is a practice that can be supported by products. You can implement it with commercial DataOps platforms, Node-RED flows or your own services, as long as mappings are versioned, owned and applied before data reaches consumers.

Does the MQTT broker perform the transformations?

No. The broker distributes messages; transformation runs in MQTT clients that subscribe to raw topics and publish modeled ones. ConnectMQ additionally decodes Sparkplug B and can republish it into ISA-95 topics.

How is industrial DataOps related to a unified namespace?

The unified namespace is where modeled data is published and discovered; DataOps is the work that makes the data in it consistent and trustworthy. A namespace without DataOps tends to become a well-organized collection of messy data.