MQTT Topic Hierarchies for Manufacturing Telemetry and OEE Workflows

Deterministic MQTT topic design is the foundational contract between edge telemetry sources and the analytics platforms that compute Overall Equipment Effectiveness (OEE). This page is part of Core Architecture & Data Mapping, and it focuses on one narrow concern: how to structure broker namespaces so that programmable logic controllers (PLCs), edge gateways, and smart actuators publish into routes that mirror physical asset boundaries while keeping subscriptions predictable. A poorly structured namespace is not a cosmetic problem — it introduces broker routing overhead, fractures telemetry lineage, and forces downstream consumers to maintain brittle mapping dictionaries that silently corrupt availability and performance numbers.

In industrial environments, topics are not merely routing keys; they are semantic namespaces that encode topology alignment, access-control boundaries, and data-lineage requirements. The hierarchy you choose on day one constrains every query, retention policy, and access rule you write for the next decade, so it must be treated as a versioned interface, not an implementation detail.

Core concept and design contract #Permalink to this section

The production-grade industrial namespace follows a strict slash-delimited progression that aligns with the ISA-95 equipment hierarchy: enterprise/site/area/line/cell/machine/component/sensor. Each segment must be bounded, lowercase-alphanumeric with underscores, and free of dynamic wildcards at the publisher level. Enterprise and site identifiers establish multi-tenant isolation; area, line, and cell segments map to physical manufacturing zones (ISA-95 levels 3 and below, mirroring the ISA-88 physical model for batch plants). Machine-level topics should correspond one-to-one with entries in the asset register, and sensor suffixes must explicitly declare measurement intent and data type.

This rigid progression is what enables hierarchical subscriptions without payload inspection. The topic plant_eu/line_04/cell_b/cnc_12/spindle/load_current immediately communicates full context, whereas a flattened convention like cnc12_load forces brokers to fan messages out to every subscriber and forces consumers to parse strings to recover topology. A subscription such as plant_eu/line_04/+/cnc_12/spindle/# then resolves to every spindle metric on cnc_12, regardless of which cell it lives in, with zero payload inspection.

Three rules form the design contract:

• Single-level wildcard (+) matches exactly one hierarchy level and is the tool for “any cell, this machine” queries. Per the MQTT 5.0 specification, + must occupy a whole level — cnc_+ is illegal.
• Multi-level wildcard (#) matches the level it occupies and all descendants, and must be the final character of a filter. Reserve it for collectors and audit consumers, never for publishers.
• Static topology in the topic, dynamic context in the payload. Identity that never changes per message (site, line, machine, metric name, data type) belongs in the path so brokers can filter on it; values, timestamps, and quality flags belong in the payload.

Metadata-in-the-path is the single most consequential decision here. Migrating from legacy Modbus or OPC UA polling into a topic tree requires translating proprietary addressing into standardized routes, and that translation is governed by PLC tag standardization, which defines deterministic mapping rules across Siemens, Allen-Bradley, and Mitsubishi controllers. Encode the data type in the suffix so consumers never guess at deserialization:

Implementation #Permalink to this section

Treat the topic as a typed object rather than a string you concatenate by hand. A small builder/parser class makes the hierarchy a single source of truth, validates segments at publish time, and gives the ingestion side a deterministic way to recover topology. The following handles the 80% case — building, validating, and parsing ISA-95-aligned topics.

from __future__ import annotations

import re
from dataclasses import dataclass, asdict

# A topic segment: lowercase alphanumerics + underscore, 1-64 chars, no wildcards.
_SEGMENT = re.compile(r"^[a-z0-9_]{1,64}$")
_LEVELS = ("enterprise", "site", "area", "line", "cell", "machine", "component", "metric")


@dataclass(frozen=True, slots=True)
class TopicAddress:
    enterprise: str
    site: str
    area: str
    line: str
    cell: str
    machine: str
    component: str
    metric: str

    def __post_init__(self) -> None:
        for level in _LEVELS:
            value = getattr(self, level)
            if not _SEGMENT.match(value):
                raise ValueError(f"invalid segment for {level!r}: {value!r}")

    def topic(self) -> str:
        """Render the fully-qualified publish topic (no wildcards allowed)."""
        return "/".join(getattr(self, level) for level in _LEVELS)

    @classmethod
    def parse(cls, topic: str) -> "TopicAddress":
        parts = topic.split("/")
        if len(parts) != len(_LEVELS):
            raise ValueError(f"expected {len(_LEVELS)} levels, got {len(parts)}: {topic!r}")
        return cls(**dict(zip(_LEVELS, parts)))

    def tags(self) -> dict[str, str]:
        """Topology as TSDB tag keys — preserves lineage end to end."""
        return asdict(self)


addr = TopicAddress(
    enterprise="plant_eu", site="hub", area="machining", line="line_04",
    cell="cell_b", machine="cnc_12", component="spindle", metric="load_current_f",
)
assert addr.topic() == "plant_eu/hub/machining/line_04/cell_b/cnc_12/spindle/load_current_f"
assert TopicAddress.parse(addr.topic()).tags()["machine"] == "cnc_12"

Because the topic carries data-type and precision hints, edge publishers should round before transmission to avoid IEEE 754 drift accumulating downstream. The rounding contract itself lives in precision and rounding limits; enforce it at the edge and normalize timestamps to UTC ISO 8601 so consumers never reconcile mixed offsets:

import decimal
from datetime import datetime, timezone


def normalize_telemetry(raw_value: float, precision: int = 2) -> dict:
    """Enforce deterministic rounding and UTC timestamp alignment at the edge."""
    ctx = decimal.Context(prec=precision + 6)
    rounded = ctx.create_decimal_from_float(raw_value).quantize(
        decimal.Decimal(10) ** -precision
    )
    return {
        "value": float(rounded),
        "ts_utc": datetime.now(timezone.utc).isoformat(timespec="milliseconds"),
        "q": 192,  # OPC-UA-style quality flag: 192 = GOOD
    }

Subscribers should map the parsed topology directly onto measurement schemas, partition keys, and retention tiers. The end-to-end ingestion pattern — batched writes, backpressure, and schema validation before persistence — is covered in time-series database sync; for high-volume plants, offload the write path to a worker pool as described in async batch processing. A minimal collector that batches by count and by interval:

import asyncio
from collections import deque

import paho.mqtt.client as mqtt


class TelemetryIngestor:
    def __init__(self, broker: str, batch_size: int = 500, flush_interval: float = 2.0):
        self.client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
        self.client.on_message = self._on_message
        self.client.connect(broker, 1883, 60)
        self.batch: deque[dict] = deque()
        self.batch_size = batch_size
        self.flush_interval = flush_interval

    def _on_message(self, client, userdata, msg) -> None:
        try:
            addr = TopicAddress.parse(msg.topic)
        except ValueError:
            client.publish(f"plant_eu/dlq/parse/{msg.topic}", msg.payload, qos=1)
            return
        self.batch.append({"tags": addr.tags(), "payload": msg.payload})
        if len(self.batch) >= self.batch_size:
            self._flush_to_tsdb()

    def _flush_to_tsdb(self) -> None:
        rows, self.batch = list(self.batch), deque()
        # write `rows` to the TSDB with retry/backoff; topology is already tagged
        ...

    async def run(self) -> None:
        # collector uses '#' deliberately; publishers never do
        self.client.subscribe("plant_eu/line_04/+/+/+/+/#", qos=1)
        self.client.loop_start()
        try:
            while True:
                await asyncio.sleep(self.flush_interval)
                if self.batch:
                    self._flush_to_tsdb()
        finally:
            self.client.loop_stop()

Edge cases and failure modes #Permalink to this section

Real factories break clean designs in predictable ways:

• Retained messages on identity topics. Setting retain=true on a high-frequency metric makes every new subscriber receive a stale value as if it were live, skewing the first OEE window after a reconnect. Reserve retained messages for slow-changing state such as .../mode or .../recipe_id, never for streaming metrics.
• PLC scan-cycle misalignment. A controller scanning at 10 ms and a publisher firing at 250 ms means most edge values are already stale samples. Stamp the timestamp at acquisition, not at publish, or downstream aggregation will attribute readings to the wrong shift. Where edge and server clocks disagree, apply clock drift correction before any time-bucketed math.
• Broker failover and duplicate events. With QoS 1 and a session that survives failover, redelivery produces duplicate messages. A naive consumer that increments a part counter on each message will overcount production and inflate the performance factor. Make discrete-event handlers idempotent (dedupe on a (topic, ts_utc) key) and pick the right delivery guarantee per stream — see best practices for MQTT QoS levels in factory networks for the per-stream decision matrix.
• Topology churn. Renaming a line or swapping a machine mid-shift orphans every historical query. Version the namespace (carry a schema_rev in the payload) and treat machine swaps as new asset-register entries rather than in-place renames.
• Dead-letter handling. When broker queues saturate or a consumer fails, dropped packets must be captured for audit and replay. Mosquitto does not natively redirect failed deliveries via topic rewrites; deploy a bridge or a dedicated monitoring consumer that republishes unacknowledged or unparseable messages to a plant_eu/dlq/# namespace after a timeout. HiveMQ’s Enterprise Dead Letter Queue extension provides native DLQ support with configurable retention. Either way, the collector above routes parse failures to the dead-letter tree instead of dropping them silently.

Verification and testing #Permalink to this section

Verify the namespace at three layers: the topic contract in unit tests, the wire in broker logs, and the result in the TSDB.

import pytest


def test_publisher_topics_carry_no_wildcards():
    addr = TopicAddress(
        enterprise="plant_eu", site="hub", area="machining", line="line_04",
        cell="cell_b", machine="cnc_12", component="spindle", metric="rpm_i",
    )
    topic = addr.topic()
    assert "+" not in topic and "#" not in topic
    # round-trip: parsing must reconstruct identical topology tags
    assert TopicAddress.parse(topic).tags() == addr.tags()


def test_rejects_uppercase_and_spaces():
    with pytest.raises(ValueError):
        TopicAddress(
            enterprise="Plant EU", site="hub", area="machining", line="line_04",
            cell="cell_b", machine="cnc_12", component="spindle", metric="rpm_i",
        )

Confirm what is actually on the wire by subscribing with a wildcard and printing topics — a fast way to catch publishers that drift from the contract:

mosquitto_sub -h broker.plant_eu -t 'plant_eu/line_04/#' -v -F '%t %p' | head

Finally, prove the topology survived ingestion by querying the TSDB on the tag keys the parser produced (TimescaleDB shown):

SELECT machine, component, metric, count(*) AS samples
FROM telemetry
WHERE site = 'hub' AND line = 'line_04'
  AND ts_utc >= now() - interval '1 hour'
GROUP BY machine, component, metric
ORDER BY samples DESC;

If a machine or component column is NULL, a publisher escaped the topic contract — fix it at the source, not with a downstream patch.

Performance and scale considerations #Permalink to this section

Topic structure directly drives broker CPU, memory, and downstream query cost. A few levers matter at plant scale:

• Subscription selectivity. Hierarchical filters let the broker’s topic-matching tree prune early; flat namespaces force near-broadcast fan-out and inflate CPU under burst. Keep depth meaningful (6–8 levels) but avoid unbounded width — thousands of sibling topics under one node slow the matching tree.
• Shared subscriptions for horizontal scale. Use MQTT 5 shared subscriptions ($share/ingest/plant_eu/line_04/#) to load-balance one logical stream across a pool of collector instances without duplicating delivery. This is the broker-native complement to a worker queue such as Celery-based high-throughput ingestion.
• Batching vs. latency. Batched writes cut TSDB write amplification during PLC burst events, but the flush interval bounds end-to-end latency. For sub-100 ms OEE refresh, size batches by count first and treat the interval as a safety net, as the collector above does.
• Retention tiering by topology. Because the parsed topic becomes tag keys, you can tier retention per component — keep spindle/vibration_rms at full resolution for condition monitoring while downsampling coolant/temp after 7 days. Map tags to partition keys so high-cardinality metrics do not bloat a single chunk.
• Cardinality discipline. Every distinct topic is a series. Putting volatile values (a batch ID, a timestamp) in the path explodes cardinality and degrades both broker matching and TSDB indexing — another reason dynamic context belongs in the payload.

Aligning topic hierarchies with physical plant topology, standardizing tag translation, enforcing edge-side rounding, and batching TSDB writes lets engineering teams reach sub-100 ms latency for OEE computation while preserving audit-grade lineage. The downstream consumer of all of this is the availability and performance math itself, governed by OEE formula validation and shift boundary logic.

Related #Permalink to this section

• Up one level: Core Architecture & Data Mapping
• PLC Tag Standardization — deterministic mapping from controller addresses to topic suffixes
• Precision & Rounding Limits — IEEE 754 drift control before transmission
• Time-Series Database Sync — partitioning and ingestion for the routed streams
• Best Practices for MQTT QoS Levels in Factory Networks — per-stream delivery-guarantee selection
• Clock Drift Correction — aligning edge timestamps before time-bucketed math

Reference the MQTT 5.0 specification for wildcard and topic-name constraints, and the Eclipse Paho Python client documentation for production-grade connection management.