Industrial Site AI Integration Architecture Guide, Part 2: Real-Time Anomaly Detection and Predictive Maintenance

Why Industrial Data Is Different

Unlike typical web data, industrial site data consists of thousands of sensor readings per second, millisecond-precision timestamps, and proprietary protocols from decades-old legacy equipment. Building on the AI-OT gateway from Part 1, this article covers the pipeline that turns this data into a form AI can consume.

MQTT → Standard Event Stream

Python

import paho.mqtt.client as mqtt
import json
from datetime import datetime, timezone

def on_message(client, userdata, msg):
    raw = json.loads(msg.payload)
    event = {
        "specversion": "1.0",
        "type": "sensor.measurement",
        "source": f"factory/line-A/{msg.topic}",
        "time": datetime.now(timezone.utc).isoformat(),
        "data": {
            "value": raw["v"],
            "unit": raw.get("u", "?"),
            "quality": raw.get("q", 192),
        }
    }
    kafka_producer.send("sensor-events", json.dumps(event).encode())

client = mqtt.Client()
client.on_message = on_message
client.connect("broker.factory.local", 1883)
client.subscribe("sensors/#", qos=1)

Edge Buffering: Handling Network Outages

Industrial site networks are unreliable. Data must not be lost even if the WAN connection drops.

Python

import sqlite3

class EdgeBuffer:
    def __init__(self, db_path: str):
        self.conn = sqlite3.connect(db_path)
        self.conn.execute("PRAGMA journal_mode=WAL")
        self.conn.execute(
            "CREATE TABLE IF NOT EXISTS events ("
            "id INTEGER PRIMARY KEY AUTOINCREMENT,"
            "payload TEXT NOT NULL,"
            "sent INTEGER DEFAULT 0)"
        )

    def write(self, payload: str):
        self.conn.execute("INSERT INTO events(payload) VALUES(?)", (payload,))
        self.conn.commit()

    def flush(self, producer, batch: int = 1000):
        rows = self.conn.execute(
            "SELECT id, payload FROM events WHERE sent=0 LIMIT ?", (batch,)
        ).fetchall()
        for _, p in rows:
            producer.send("sensor-events", p.encode())
        producer.flush()
        ids = [str(r[0]) for r in rows]
        self.conn.execute(
            f"UPDATE events SET sent=1 WHERE id IN ({','.join(ids)})"
        )
        self.conn.commit()

TimescaleDB Design

SQL

CREATE TABLE sensor_readings (
    time        TIMESTAMPTZ NOT NULL,
    device_id   TEXT NOT NULL,
    metric      TEXT NOT NULL,
    value       DOUBLE PRECISION
);

SELECT create_hypertable('sensor_readings', 'time',
    chunk_time_interval => INTERVAL '1 hour');

-- (7 after)
SELECT add_compression_policy('sensor_readings', INTERVAL '7 days');

-- 1
CREATE MATERIALIZED VIEW sensor_1min
WITH (timescaledb.continuous) AS
SELECT time_bucket('1 minute', time) AS bucket,
       device_id, metric,
       AVG(value) avg_val, MAX(value) max_val
FROM sensor_readings
GROUP BY bucket, device_id, metric;

Kafka vs AWS IoT Greengrass

Aspect	Apache Kafka	AWS IoT Greengrass
Setup complexity	High	Low (managed)
Offline buffering	Implement yourself	Built-in
Edge ML inference	Separate setup	Lambda integration
Cost	Infrastructure cost	Usage-based billing

End-to-End Pipeline

CODE

OPC-UA / MQTT / Modbus
    ↓
Edge gateway (EdgeBuffer + protocol conversion)
    ↓ Kafka
Stream processing (Kafka Streams)
    ↓
TimescaleDB (short-term high-resolution) + S3 (long-term cold)
    ↓
AI models (anomaly detection / predictive maintenance / quality inspection)

Part 3 will cover deploying predictive maintenance models that run in real time on this pipeline, along with drift detection.

Diagnosing and Fixing Missing or Duplicate Sensor Data

Many anomaly-detection false alarms come not from the model but from missing, duplicated, or mis-timestamped input.

Causes

Publishing with MQTT QoS 0 loses messages during brief network drops
QoS 1 is "at least once," so retransmissions store duplicates
Gateway clocks drift (no NTP sync) and timestamps arrive out of order

Reproduce and verify

Bash

# watch topic flow with QoS
mosquitto_sub -h broker.local -t 'plant/+/sensor/#' -q 1 -v
# gateway clock sync state
chronyc tracking

SQL

-- TimescaleDB: find gaps in 1-minute buckets
SELECT time_bucket_gapfill('1 minute', ts) AS b, device_id, count(*) AS n
FROM sensor_readings
WHERE ts > now() - interval '6 hours' AND ts < now()
GROUP BY b, device_id
HAVING count(*) = 0;
-- duplicates
SELECT device_id, ts, count(*) FROM sensor_readings
GROUP BY device_id, ts HAVING count(*) > 1 LIMIT 20;

Fix

Publish critical signals with QoS 1 or higher and make storage idempotent: a unique constraint on (device_id, ts) plus INSERT ... ON CONFLICT DO NOTHING.
Timestamp at measurement on the gateway (not on receipt) and monitor sync state with chrony.
Don't interpolate gaps into model input; pass a "missing" flag so missing data isn't mistaken for an anomaly.

Prevention

Track per-device receive rate (actual vs expected count) on the dashboard and fire a data-quality alert first when it drops below threshold.

#IIoT

✦ ✦ ✦

Sources and verification

Sources and checks vary by article. Only articles with a recorded check date, scope and sources below were checked, and only within that scope. For other articles, verify against the current docs for your version before applying anything, and email us about errors. Editorial policy · email us

This article has no recorded human check (date and scope) yet.

Published · May 21, 2026

How we write and review →Editorial policy →

Was this helpful?

Comments

Be the first to comment.