Payload Serialization Formats for Manufacturing MQTT Telemetry

Serialization is the layer of Core Architecture & Data Mapping that decides what actually crosses the wire once a MQTT topic hierarchy has told a message where it belongs. This guide covers one narrow but consequential decision: how to encode a sensor reading into bytes so that the broker, every downstream consumer, and the time-series database agree on its meaning without inspecting a human-readable blob every time. Picking a format is not a stylistic preference. It sets the bandwidth budget for a cellular or 4G edge uplink, determines whether a firmware update can add a field without breaking every existing subscriber, and decides whether a NaN from a failed transducer round-trips cleanly or corrupts the batch. Teams that treat serialization as an afterthought discover the cost only after thousands of edge nodes are already shipping a format that cannot evolve.

Why this matters more on a factory floor than in a typical web API: manufacturing MQTT fleets publish continuously, at high fan-out, over links that are frequently metered or congested, and the consumers on the other end include long-lived historians that must still decode a payload written by firmware from three years ago. A format that is cheap to write and easy to debug in a browser (JSON) is not automatically the right choice once a plant has ten thousand publishing sensors and a schema that changes every quarter.

Format comparison at a glance #Permalink to this section

The table below is the decision reference this whole topic builds on. “Typical size” is measured for the same representative record used in the diagram above — an asset ID, a metric name, a millisecond timestamp, a double-precision value, a quality flag, and a sequence number.

JSON’s size disadvantage looks small per message, but it compounds: at 50,000 messages per second across a plant, the difference between a 120-byte JSON record and a 35-byte Protobuf record is roughly 4 MB/s of sustained broker throughput — bandwidth that matters on a shared cellular backhaul or a congested plant VLAN. The comparison of JSON, Protobuf and Avro works through that measurement in detail; designing Sparkplug B topic namespaces covers the birth/death lifecycle that gets Sparkplug down to its smallest steady-state size.

Core concept and design contract #Permalink to this section

A wire-format contract has to answer four questions independently of which encoding you pick: how does a consumer know which schema produced these bytes, how does the encoding handle values a sensor legitimately cannot report, how is the schema allowed to change without breaking a consumer that has not redeployed, and how are numeric types represented so that IEEE 754 double-precision floats from a data historian and the values that landed in the time-series column agree bit-for-bit on round-trip.

Schema identity travels with the payload, never only in the topic. It is tempting to let the MQTT topic hierarchy imply the schema — plant_eu/line_04/cnc_12/spindle/vibration_rms looks self-explanatory. But topics are static routing keys; they cannot carry a schema version. A firmware rollout that changes vibration_rms from a scalar to an RMS-plus-peak pair on the same topic silently breaks every consumer unless the payload itself declares which schema produced it — a header byte, a Confluent-style four-byte schema ID, or (for Sparkplug B) the alias assigned at NBIRTH.

Sensor-fault values need an explicit representation, not a special float. IEEE 754 defines NaN and ±Infinity, and Protobuf and Avro’s binary float/double types can carry them natively because they encode the raw IEEE 754 bit pattern. JSON, per RFC 8259, has no token for either — strict parsers reject a literal NaN in the token stream. Python’s json module will emit NaN by default (a non-standard extension) and then fail to parse it in a strict downstream consumer, which is exactly the kind of asymmetric bug that only shows up in production. The design contract is to never serialize NaN/Infinity as a value at all: carry a quality flag (GOOD/UNCERTAIN/BAD) alongside the numeric field and set the value to 0.0 or omit the field, following the same quality-flag discipline used across ingestion and cleaning workflows.

Schema evolution is additive by contract, not by convention. Every format here supports adding a field without breaking old consumers, but only if you reserve retired field numbers (Protobuf, Sparkplug B) or ship default values for new fields (Avro). This is developed fully in schema evolution for versioned sensor payloads; the constraint that matters here is that the encoding you choose must make violating additivity structurally awkward, not merely discouraged in a wiki page.

Numeric representation is decided once, upstream of the encoding. Whether a counter accumulates in float32, float64, or a fixed-point integer is a decision made at the precision and rounding boundary, before serialization. The serialization layer’s job is to preserve whatever precision that boundary decided on — Protobuf’s double and Avro’s double are both full 64-bit IEEE 754, so neither introduces additional loss; a naive hand-rolled binary format using struct.pack("f", ...) silently truncates to 32-bit and re-introduces the drift documented in handling floating-point drift in sensor readings.

Implementation #Permalink to this section

The reference implementation below is a minimal schema registry and Avro codec that mirrors the Confluent wire-format convention (a magic byte plus a four-byte big-endian schema ID prefixed to the binary body) but is broker-agnostic — it works equally well prefixed onto an MQTT retained message as it does inside a Kafka record. Avro is the right format to show a registry against because its reader/writer schema resolution model makes the versioning contract explicit rather than implicit.

from __future__ import annotations

import io
import struct
from dataclasses import dataclass
from typing import Any

import fastavro

MAGIC_BYTE = b"\x00"  # Confluent-style wire-format marker


@dataclass(frozen=True, slots=True)
class SchemaEntry:
    schema_id: int
    parsed_schema: dict[str, Any]


class MqttSchemaRegistry:
    """In-memory registry keyed by (subject, version).

    A subject is the canonical metric family behind an MQTT topic, e.g.
    'plant_eu/line_04/cnc_12/spindle/vibration_rms'. Production deployments
    back this with a durable store (Redis, Postgres) shared by every edge
    publisher and every consumer, so a schema ID minted on one gateway
    resolves identically on every historian that reads it.
    """

    def __init__(self) -> None:
        self._by_id: dict[int, SchemaEntry] = {}
        self._by_subject_version: dict[tuple[str, int], int] = {}
        self._next_id = 1

    def register(self, subject: str, version: int, schema: dict[str, Any]) -> int:
        parsed = fastavro.parse_schema(schema)
        schema_id = self._next_id
        self._next_id += 1
        self._by_id[schema_id] = SchemaEntry(schema_id, parsed)
        self._by_subject_version[(subject, version)] = schema_id
        return schema_id

    def encode(self, subject: str, version: int, record: dict[str, Any]) -> bytes:
        schema_id = self._by_subject_version[(subject, version)]
        entry = self._by_id[schema_id]
        buf = io.BytesIO()
        buf.write(MAGIC_BYTE)
        buf.write(struct.pack(">I", schema_id))  # big-endian, matches Confluent wire format
        fastavro.schemaless_writer(buf, entry.parsed_schema, record)
        return buf.getvalue()

    def decode(self, payload: bytes) -> dict[str, Any]:
        if payload[:1] != MAGIC_BYTE:
            raise ValueError("payload missing schema-registry magic byte")
        schema_id = struct.unpack(">I", payload[1:5])[0]
        entry = self._by_id.get(schema_id)
        if entry is None:
            # Unknown schema_id: the consumer's registry cache is stale,
            # or the producer registered against a different registry
            # instance entirely. Route to DLQ rather than guess a schema.
            raise KeyError(f"unknown schema_id={schema_id}; registry not synced")
        buf = io.BytesIO(payload[5:])
        return fastavro.schemaless_reader(buf, entry.parsed_schema)


TELEMETRY_SCHEMA_V1 = {
    "type": "record",
    "name": "TelemetryRecord",
    "namespace": "plant_eu.telemetry",
    "fields": [
        {"name": "asset_id", "type": "string"},
        {"name": "metric_name", "type": "string"},
        {"name": "ts_unix_ms", "type": "long"},
        {"name": "value", "type": "double"},
        {"name": "quality", "type": {"type": "enum", "name": "Quality",
                                      "symbols": ["GOOD", "UNCERTAIN", "BAD"]}},
        {"name": "seq", "type": "int"},
    ],
}

registry = MqttSchemaRegistry()
registry.register("plant_eu/line_04/cnc_12/spindle/vibration_rms", 1, TELEMETRY_SCHEMA_V1)

For payloads where the schema is stable, is owned entirely by one engineering team, and does not need registry-mediated resolution, Protobuf is the leaner choice. The schema lives in a .proto file compiled with protoc, and the generated Python bindings do the encoding:

# telemetry.proto — compiled via `protoc --python_out=. telemetry.proto`
PROTO_SCHEMA = '''
syntax = "proto3";
package plant_eu.telemetry;

message TelemetryRecord {
  string asset_id = 1;
  string metric_name = 2;
  int64 ts_unix_ms = 3;
  double value = 4;
  Quality quality = 5;
  uint32 seq = 6;

  enum Quality {
    GOOD = 0;
    UNCERTAIN = 1;
    BAD = 2;
  }

  reserved 7, 8;               // retired field numbers, never reused
  reserved "raw_unit_code";    // retired field name, likewise reserved
}
'''

from plant_eu.telemetry import telemetry_pb2  # generated by protoc


def encode_protobuf(asset_id: str, metric: str, ts_ms: int, value: float,
                     quality: "telemetry_pb2.TelemetryRecord.Quality", seq: int) -> bytes:
    """Encode one telemetry record to its compact binary wire form."""
    msg = telemetry_pb2.TelemetryRecord(
        asset_id=asset_id, metric_name=metric, ts_unix_ms=ts_ms,
        value=value, quality=quality, seq=seq,
    )
    return msg.SerializeToString()


def decode_protobuf(payload: bytes) -> "telemetry_pb2.TelemetryRecord":
    """Decode; raises google.protobuf.message.DecodeError on malformed bytes.

    Unknown fields from a newer producer schema are preserved, not
    rejected — proto3's forward-compatibility guarantee — so a consumer
    running an older .proto still parses records from a newer publisher.
    """
    msg = telemetry_pb2.TelemetryRecord()
    msg.ParseFromString(payload)
    return msg

Edge cases and failure modes #Permalink to this section

• Schema drift between producer and consumer. An edge gateway ships a new firmware image with schema version 3 before the historian has deployed the matching decoder. With Avro and the registry pattern above, the consumer’s KeyError on an unrecognized schema_id is the correct failure — route the batch to a dead-letter table rather than guessing. With Protobuf, proto3’s forward-compatible wire format means an old consumer silently drops fields it does not recognize instead of erroring, which is safer for uptime but means you must monitor for “field never populated” as its own signal, since there is no exception to catch.
• Endianness in hand-rolled binary formats. Protobuf and Avro both define their wire encoding precisely (varints are inherently endian-neutral; Avro’s fixed-width types are big-endian for the fixed logical type but most numeric types use variable-length zig-zag encoding), so cross-platform decode is safe by construction. The risk appears when a team “optimizes” a hot path with a custom struct.pack binary format and forgets to pin byte order — struct.pack("f", value) uses native byte order, which differs between an ARM-based edge gateway and an x86 ingestion server. Always pin it explicitly: struct.pack("<f", value) for little-endian, struct.pack(">f", value) for big-endian, and pick one for the whole fleet.
• NaN and Infinity crossing a JSON boundary. A quality-flagged BAD reading from a disconnected thermocouple must never be encoded as a raw NaN token if any hop in the pipeline uses a strict JSON parser (most non-Python consumers, and Python’s own json.loads with parse_constant overridden). Encode the fault as quality: "BAD" with a sentinel or omitted value field, never as the IEEE 754 special value itself, in any format — Protobuf and Avro can carry a literal NaN correctly, but a NaN in a TSDB column breaks AVG()/SUM() aggregations silently in most SQL engines.
• Decimal precision surviving the round-trip. A cumulative production counter serialized as Protobuf double or Avro double preserves full 64-bit precision, but if the source value was already accumulated in float32 upstream, serialization cannot recover the lost precision — it only avoids adding more. See Decimal vs float for cumulative production counters for where that decision has to be made.
• Payload size against QoS 2’s in-flight window. A verbose JSON payload combined with QoS 2 delivery multiplies broker session-state cost — each in-flight message holds state for the full handshake duration. Keep QoS 2 payloads compact regardless of format; the message_size_limit guidance in MQTT QoS levels for factory networks is a hard ceiling, not a target.

Verification and testing #Permalink to this section

Correctness has two layers: a deterministic round-trip test per format, and a wire-level inspection that confirms what the broker is actually carrying.

def test_avro_roundtrip_is_lossless():
    registry = MqttSchemaRegistry()
    registry.register("test/subject", 1, TELEMETRY_SCHEMA_V1)

    record = {
        "asset_id": "cnc-12", "metric_name": "vibration_rms", "ts_unix_ms": 1_752_000_000_123,
        "value": 0.842, "quality": "GOOD", "seq": 4711,
    }
    payload = registry.encode("test/subject", 1, record)
    assert registry.decode(payload) == record


def test_unknown_schema_id_routes_to_dlq_not_a_guess():
    registry = MqttSchemaRegistry()
    registry.register("test/subject", 1, TELEMETRY_SCHEMA_V1)
    forged = b"\x00" + (99999).to_bytes(4, "big") + b"\x00"
    try:
        registry.decode(forged)
        assert False, "expected KeyError on unregistered schema_id"
    except KeyError:
        pass

Inspecting the wire confirms the encoding matches the contract rather than trusting the publisher’s code path. Subscribe with mosquitto_sub and print raw payload length per message to spot an accidental fallback to JSON on a topic that should carry Protobuf:

mosquitto_sub -h broker.plant_eu -t 'plant_eu/line_04/+/+/+/vibration_rms_pb' \
  -F '%l bytes @ %t' -v | head -n 20

And confirm, in the time-series database, that decode failures are visible rather than silently dropped — a rising count here is an early signal of a firmware rollout shipping an unregistered schema:

SELECT schema_id, count(*) AS rejected, min(received_at) AS first_seen
FROM telemetry_dlq
WHERE reject_reason = 'unknown_schema_id'
  AND received_at >= now() - interval '1 hour'
GROUP BY schema_id
ORDER BY rejected DESC;

Performance and scale considerations #Permalink to this section

At representative record sizes, the throughput gap between formats is a function of both bytes-on-wire and CPU per decode, and the two do not always favor the same format. JSON’s text parsing cost scales with string length and grows with UTF-8 validation overhead; Protobuf and Avro decode against a fixed binary layout and a C-extension parser (fastavro’s Cython path, protobuf’s C++ backend via upb), which is typically 5–10x faster per record at batch scale even before accounting for the smaller payload.

• Bytes per message compounds at fleet scale. The ~85-byte gap between JSON and Protobuf for the representative record above becomes roughly 4 MB/s of sustained broker throughput at 50,000 msg/s — the difference between comfortably fitting a plant’s WAN uplink and saturating it during a shift-change burst.
• Batch decode cost dominates ingestion CPU, not the network. At sustained high-volume ingestion, profile the decode step separately from the broker read; a naive per-message fastavro.schemaless_reader call that re-parses the schema string on every invocation (instead of caching the parse_schema() result, as the registry above does) can dwarf the actual I/O cost.
• Schema-registry lookups must be cached, not queried per message. The registry pattern above resolves schema_id to a parsed schema from an in-process cache; a design that hits a remote registry service per message adds a network round trip to every decode and becomes the bottleneck long before the TSDB write does.
• Cardinality discipline extends to schema versions. If schema_version is stored as a tag/label alongside asset_id and metric_name, an unbounded number of live schema versions inflates series cardinality the same way an unbounded topic namespace does. Deprecate old schema versions on a defined cadence once consumers have migrated, consistent with the time-series database sync guidance on cardinality.
• Async batch writers amortize the decode cost. Decoding and validating in the same async batch processing worker pool that handles TSDB writes avoids a second queueing hop; for very high fan-in topics, Celery-based MQTT ingestion distributes decode load the same way it distributes write load.

Getting the format right at the start is cheaper than migrating a fleet later: once ten thousand edge nodes are shipping a schema, evolving it safely — rather than replacing it — becomes the operative concern, which is exactly what schema evolution for versioned sensor payloads addresses.

Related #Permalink to this section

• JSON vs Protobuf vs Avro for MQTT Sensor Payloads — measured byte sizes and decode throughput for the three general-purpose formats
• Schema Evolution for Versioned Sensor Payloads — additive-only field rules and a CI compatibility gate
• Designing Sparkplug B Topic Namespaces for Factory Telemetry — the MQTT-native profile that pairs a fixed protobuf schema with alias maps
• MQTT Topic Hierarchies — the routing layer this payload rides inside
• Up: Core Architecture & Data Mapping