Industrial Time Series Data: Sampling, Compression and Storage

Industrial time series data is the stream of timestamped measurements that machines, sensors and controllers produce every second of every shift. Handling it well means choosing how often to sample, what to throw away, how to timestamp it and where to keep it. This article walks through each of those decisions and gives you working code to apply deadband compression on MQTT data before it reaches a historian.

What Makes Industrial Time Series Data Different

All time series share the same shape: a timestamp, an identifier and a value. Industrial data adds several complications that IT monitoring data rarely has.

• Very different change rates. A tank level changes slowly over hours; a vibration signal changes thousands of times per second. One sampling policy cannot fit both.
• Long retention requirements. Quality investigations, regulatory records and model training often need years of history, not weeks.
• Quality matters as much as value. A value from a disconnected sensor is not a measurement. Industrial systems carry a quality flag alongside every point.
• Event and state data mixed with analog data. Machine states, alarms, batch IDs and recipe changes are sparse and must be joined with dense analog signals.
• Clocks you do not control. PLCs, gateways and servers each have their own clock, and they drift.

Getting these right at the start is far cheaper than repairing years of inconsistent history later.

Choosing Sampling Rates

The sampling rate should follow the physics of the signal and the question you want to answer, not the default scan time of the PLC.

Two rules help. First, the Nyquist principle: to see a variation, sample at least twice as fast as that variation, and in practice several times faster. Second, do not stream raw high-frequency waveforms through MQTT continuously. Compute features at the edge and publish the features; send raw snapshots only on demand or on events.

For discrete data, report by exception is the natural pattern. The report by exception article explains how it reduces traffic without losing information.

Compression: Keeping the Signal, Dropping the Noise

Compression in industrial historians usually means lossy filtering at ingestion, not zip files. The goal is to store fewer points while being able to reconstruct the signal within a known tolerance.

Deadband (exception) filtering

The simplest method: store a new point only when the value differs from the last stored value by more than a threshold. A deadband of 0.2 degC on a temperature means fluctuations below 0.2 degC are not stored. Most implementations also store a point after a maximum time even if nothing changed (a heartbeat), so consumers can tell a steady signal from a dead one.

Swinging door compression

Swinging door trending, used in many commercial historians, keeps a point only when a straight line from the last stored point can no longer represent the intermediate values within the tolerance. It preserves slopes much better than a plain deadband, so ramps are stored with a few points instead of many.

Choosing tolerances

Set the tolerance relative to the sensor's accuracy, not to an arbitrary percentage. Filtering below the sensor's noise floor stores noise; filtering far above it loses real process changes. Document the tolerance per tag type in your IIoT data model so it is reviewed like any other model attribute.

Deadband Compression on MQTT Data in Python

This subscriber reads JSON telemetry from MQTT, applies a deadband with a heartbeat, and writes surviving points as InfluxDB line protocol to standard output, which you can redirect to a file or pipe to a writer. It uses the free public MQTT broker with a demo simulator built in.

import json
import math
import random
import threading
import time
import uuid

import paho.mqtt.client as mqtt

BROKER = "broker.connectmq.com"
PREFIX = f"connectmq/demo/{uuid.uuid4().hex[:8]}/plant1/line1"
DEADBAND = {"temperature": 0.2, "pressure": 0.05}
HEARTBEAT_S = 60

last_stored = {}  # topic -> (value, ts_seconds)


def keep(topic, metric, value, ts):
    prev = last_stored.get(topic)
    if prev is None:
        return True
    band = DEADBAND.get(metric, 0.0)
    return abs(value - prev[0]) > band or ts - prev[1] >= HEARTBEAT_S


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


def on_message(client, userdata, msg):
    asset, metric = msg.topic.split("/")[-2:]
    p = json.loads(msg.payload)
    ts = p["ts"] / 1000
    if keep(msg.topic, metric, p["value"], ts):
        last_stored[msg.topic] = (p["value"], ts)
        print(f'{metric},asset={asset},quality={p["quality"]} value={p["value"]} {int(p["ts"] * 1e6)}')


def simulate(pub):
    t0 = time.time()
    while True:
        t = time.time() - t0
        temp = 60 + 5 * math.sin(t / 120) + random.gauss(0, 0.05)
        pres = 2.5 + random.gauss(0, 0.01)
        ts = int(time.time() * 1000)
        for metric, val in [("temperature", temp), ("pressure", pres)]:
            body = {"value": round(val, 3), "ts": ts, "quality": "GOOD"}
            pub.publish(f"{PREFIX}/pump1/{metric}", json.dumps(body), qos=0)
        time.sleep(1)


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="tsdb-" + uuid.uuid4().hex[:8])
client.on_connect = on_connect
client.on_message = on_message
client.connect(BROKER, 1883, keepalive=60)
threading.Thread(target=simulate, args=(client,), daemon=True).start()
client.loop_forever()

Run it for a few minutes and count lines: the pressure, which only shows noise below its deadband, produces roughly one point per minute (the heartbeat), while the temperature produces points as the sine wave moves. That is exactly the behaviour you want from a historian: dense where the process moves, sparse where it does not.

Timestamps and Ordering

Always timestamp at the source, as close to the measurement as possible, in UTC with milliseconds. Broker arrival time is not measurement time, especially when an edge gateway buffers data during a network outage and forwards it later. The store and forward pattern relies on source timestamps to put late data in the right place.

Synchronize clocks with NTP or PTP on gateways and, where possible, on controllers. Expect out-of-order arrival and make sure your storage accepts writes into the past. Some time series databases handle late writes efficiently; others penalize them.

Storage Options for Industrial Time Series Data

Many plants combine a historian or TSDB for recent data with a lake for long-term analytics. The MQTT historian article walks through building the historian side directly from MQTT topics.

Getting Data from the Broker into Storage

A broker is a router, not a database: messages flow through it and retained messages hold only the last value per topic. To store history you run a writer that subscribes and inserts. Telegraf is a common choice:

[[inputs.mqtt_consumer]]
  servers = ["ssl://broker.connectmq.com:8883"]
  topics = ["connectmq/plant1/#"]
  client_id = "YOUR_CLIENT_ID"
  username = "YOUR_USERNAME"
  password = "YOUR_PASSWORD"
  qos = 1
  data_format = "json"
  json_time_key = "ts"
  json_time_format = "unix_ms"
  tag_keys = ["quality"]

  [[inputs.mqtt_consumer.topic_parsing]]
    topic = "connectmq/+/+/+/+"
    tags = "_/site/line/asset/_"
    measurement = "_/_/_/_/measurement"

[[outputs.influxdb_v2]]
  urls = ["http://localhost:8086"]
  token = "YOUR_INFLUX_TOKEN"
  organization = "plant"
  bucket = "telemetry"

With ConnectMQ as the managed MQTT broker, Telegraf is simply another authenticated client over TLS on port 8883. Use QoS 1 for the writer subscription so a brief disconnect does not silently drop data, and keep the writer's session persistent if you want queued messages while it restarts.

Checklist for Industrial Time Series Data

1. Classify each signal: analog slow, analog fast, discrete, or high-frequency.
2. Set sampling and deadband per class, based on sensor accuracy.
3. Add a heartbeat so steady signals are distinguishable from dead ones.
4. Timestamp at the source in UTC with milliseconds.
5. Publish quality with every value.
6. Compute features at the edge for kHz signals.
7. Choose hot and cold storage, and define retention and downsampling.
8. Test late and out-of-order writes before go-live.

For the architecture around all of this, see the industrial IoT with MQTT guide.

Get Started with ConnectMQ

Create a free Serverless broker at app.connectmq.com, no credit card required, then connect your gateway and a Telegraf or Python writer with your generated credentials. The ConnectMQ live topic explorer helps you check publish rates and payloads before you commit to a sampling and compression policy.

Frequently Asked Questions

What is industrial time series data?

It is timestamped measurement and event data produced by industrial equipment, such as temperatures, pressures, speeds, states and counters. It is usually stored with a quality flag and kept for long periods for operations, quality and analytics.

What is a good deadband value?

Start near the sensor's accuracy or noise level for that measurement, then check that important process changes are still visible in the stored data. Document the value per tag type and review it when sensors change.

Can an MQTT broker store time series history?

No. A broker routes messages and keeps at most the last retained value per topic. Store history with a subscriber such as Telegraf, a historian connector or your own writer.

Should I stream raw vibration data over MQTT?

Usually not continuously. Compute features like RMS, peak and band energies at the edge and publish those, then send raw waveform snapshots only on request or when an event triggers them.