IoT MLOps: Monitoring Machine Learning Models in the Field over MQTT

IoT MLOps is the practice of deploying, monitoring and updating machine learning models that run on devices you cannot see. In this guide you will build a lightweight monitoring loop: edge devices publish inference metrics over MQTT, a central Python service measures drift, and degraded models trigger alerts on a dedicated topic.

Why Models in the Field Degrade

A model that scored well in the lab starts losing accuracy the day it ships. In IoT the reasons are physical and often slow:

• Sensor aging and recalibration. A vibration sensor drifts, a camera lens gets dusty, a microphone picks up a new fan.
• Process changes. A new material, a different shift pattern or a new product variant shifts the input distribution.
• Seasonality. Ambient temperature and humidity change sensor baselines over months.
• Deployment mismatch. Quantization or a different firmware build changes outputs slightly compared to the evaluation model.

In a web service you can log every request and compare predictions to user feedback. In the field you usually have no labels at all, limited bandwidth and devices that go offline. That is why IoT MLOps leans on proxy signals: input statistics, output distributions, confidence and latency.

What to Measure on the Device

Sending raw sensor windows to the cloud for every inference defeats the point of edge AI. Instead, each device aggregates a few numbers per interval and publishes a compact metrics message.

A metrics payload might look like this:

{
  "device": "press-07",
  "model": "vib-classifier",
  "modelVersion": "1.4.2",
  "window": "2026-10-16T10:00:00Z/PT1M",
  "inferences": 600,
  "inputMean": {"rms": 0.42, "peakHz": 118.0},
  "inputStd": {"rms": 0.05, "peakHz": 6.1},
  "classCounts": {"normal": 588, "imbalance": 9, "bearing": 3},
  "confMean": 0.91,
  "confP10": 0.74,
  "latencyMsP95": 14.2
}

Topic Design for IoT MLOps

Keep model telemetry separate from process telemetry so dashboards and consumers can subscribe precisely:

connectmq/plant1/ml/<model>/<device>/metrics
connectmq/plant1/ml/<model>/<device>/status        (retained: version, loaded at)
connectmq/plant1/ml/<model>/alerts
connectmq/plant1/ml/<model>/<device>/cmd           (rollout or rollback commands)

Retaining the status topic means any new subscriber immediately learns which model version every device runs, which is the most common question during an incident.

Publishing Metrics from an Edge Device

The script below simulates an edge device running a classifier and publishing one metrics message per interval. On real hardware you would accumulate the statistics inside your inference loop. It uses the public test area of broker.connectmq.com; replace the prefix with your own random ID.

import json
import random
import time
import uuid

import paho.mqtt.client as mqtt

PREFIX = "connectmq/demo/mlops-91bc/plant1/ml/vib-classifier"
DEVICE = "press-07"

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

status = {"modelVersion": "1.4.2", "loadedAt": time.time()}
client.publish(f"{PREFIX}/{DEVICE}/status", json.dumps(status), qos=1, retain=True)

drift = 0.0
while True:
    drift += 0.002  # slowly shift the input to simulate sensor aging
    rms = [random.gauss(0.42 + drift, 0.05) for _ in range(600)]
    conf = [min(1.0, max(0.0, random.gauss(0.91 - drift, 0.06))) for _ in range(600)]
    conf.sort()
    msg = {
        "device": DEVICE,
        "modelVersion": "1.4.2",
        "inferences": len(rms),
        "inputMean": {"rms": sum(rms) / len(rms)},
        "inputHist": {"rms": [sum(1 for v in rms if lo <= v < lo + 0.05)
                              for lo in [0.25 + 0.05 * i for i in range(10)]]},
        "confMean": sum(conf) / len(conf),
        "confP10": conf[len(conf) // 10],
        "ts": time.time(),
    }
    client.publish(f"{PREFIX}/{DEVICE}/metrics", json.dumps(msg), qos=1)
    time.sleep(10)

The histogram with fixed bins is what makes drift measurable centrally without shipping raw data.

Detecting Drift Centrally

The monitor subscribes to all devices, compares each incoming histogram with a reference histogram captured at deployment, and computes the Population Stability Index (PSI). A common rule of thumb treats PSI below 0.1 as stable, 0.1 to 0.25 as worth watching and above 0.25 as a significant shift, but tune those bands on your own data.

import json
import math
import uuid

import paho.mqtt.client as mqtt

PREFIX = "connectmq/demo/mlops-91bc/plant1/ml/vib-classifier"
reference = {}  # device -> reference histogram
CONF_FLOOR = 0.70


def psi(expected, actual, eps=1e-4):
    e_total, a_total = sum(expected) or 1, sum(actual) or 1
    score = 0.0
    for e, a in zip(expected, actual):
        e_p = max(e / e_total, eps)
        a_p = max(a / a_total, eps)
        score += (a_p - e_p) * math.log(a_p / e_p)
    return score


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


def on_message(client, userdata, msg):
    m = json.loads(msg.payload)
    dev, hist = m["device"], m["inputHist"]["rms"]
    if dev not in reference:
        reference[dev] = hist
        print(f"reference captured for {dev}")
        return
    score = psi(reference[dev], hist)
    status = "ok"
    if score > 0.25 or m["confP10"] < CONF_FLOOR:
        status = "degraded"
    elif score > 0.1:
        status = "watch"
    print(f"{dev} psi={score:.3f} confP10={m['confP10']:.2f} -> {status}")
    if status != "ok":
        alert = {"device": dev, "modelVersion": m["modelVersion"],
                 "psi": round(score, 3), "confP10": round(m["confP10"], 3),
                 "status": status}
        client.publish(f"{PREFIX}/alerts", json.dumps(alert), qos=1)


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                     client_id=f"mlops-monitor-{uuid.uuid4().hex[:6]}")
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_forever()

Run both scripts and you will see the status move from ok to watch to degraded as the simulated drift accumulates. In production the reference histogram should come from the validation data used to approve the model, stored alongside the model artifact, not from the first message after boot.

Acting on Alerts

An alert is only useful if someone or something responds. Options, from simplest to most automated:

1. Notify a person. A Node-RED flow or a small service subscribes to the alerts topic and sends email or chat messages.
2. Visual alerting in the broker portal. On a private ConnectMQ broker you can define threshold alert rules on a numeric field such as psi and review hits in the alerts log.
3. Collect labels. When a device drifts, ask it to upload a few raw samples on a separate topic so engineers can label them and retrain.
4. Roll back. Publish a command on the device's cmd topic to load the previous model version, then confirm via the retained status topic.

The rollout side, how models get onto devices in the first place, is covered in deploying ML models to edge devices.

Canary Rollouts and Shadow Models

Monitoring tells you when a model degrades; rollout discipline keeps a new model from making things worse. Two patterns work well with MQTT topics.

Canary rollout. Send the new model to a small group of devices first, for example five percent of the fleet or one line per plant. Because every metrics message carries modelVersion, the monitor can compare confidence, class distribution and drift between the canary group and the rest over the same period. Promote only when the canary looks at least as healthy as the incumbent, and roll back with the same command topic if it does not.

Shadow mode. Run the new model alongside the current one on the same device, but act only on the current model's output. Publish both predictions in the metrics message, or publish a running agreement rate between the two. A shadow model that disagrees with the production model on a large share of inputs deserves a closer look before it ever takes over, even if both models seem confident.

Both patterns need nothing from the broker beyond reliable delivery and clear topics. They also make post-incident analysis easier: when an alert fires, the first question is whether it correlates with a version change, and the retained status topics answer that instantly.

A practical note on bandwidth: a metrics message every minute from each device is tiny, but raw sample uploads for labeling are not. Keep uploads on a separate topic, trigger them only on request, and stay under the message size limit of your plan.

IoT MLOps Checklist

1. Every message carries the model name and version.
2. Devices publish aggregated input statistics, not raw data, at a fixed interval.
3. A reference distribution is stored with each released model.
4. Drift and confidence thresholds are documented and reviewed after each incident.
5. Alerts go to a topic, and at least one consumer turns them into human notifications.
6. Rollback is a tested command, not a plan.
7. Retraining data is collected deliberately when drift occurs.
8. Metrics and alerts travel over TLS with per-device credentials in production.

Infrastructure for Fleet-Wide Monitoring

Metrics from hundreds of devices, a monitor, dashboards and rollout tooling all share one message bus. Running that bus yourself means patching, certificates and capacity planning. A fully managed MQTT service removes that work, and a cloud MQTT broker is reachable from devices on any network. Shared subscriptions on the paid Serverless plan let you run several monitor instances that split the metrics load.

The same metrics stream also feeds model-specific work such as predictive maintenance with machine learning, and drift checks are a close cousin of IoT anomaly detection: one watches the machine, the other watches the model. For the bigger picture of edge and cloud AI, see the AI and IoT guide.

Monitor Your Field Models with ConnectMQ

Create a private broker on the free Serverless plan at app.connectmq.com, no credit card required, point your devices at it with the generated credentials over port 8883, and watch model metrics arrive in the live topic explorer. Adding the drift monitor is the script above with a username and password.

Frequently Asked Questions

How is IoT MLOps different from regular MLOps?

Field devices rarely have labels, have limited bandwidth and can be offline, so monitoring relies on aggregated input and output statistics rather than per-request logs. Deployment and rollback must also work over unreliable links.

How often should devices publish model metrics?

Once per minute or per few hundred inferences is a good starting point. It is frequent enough to catch drift within hours and small enough to stay well within message rate limits.

What drift metric should I use?

Population Stability Index and Kolmogorov-Smirnov tests are common for inputs, and tracking low-percentile confidence catches many output problems. Use more than one signal, since each misses some failure modes.

Can the broker detect drift for me?

No. The broker transports the metrics; drift detection runs in your own service that subscribes to them. Threshold alert rules in the ConnectMQ portal can flag simple numeric limits once your service publishes a drift score.