MQTT Data Historian: Capturing Industrial Time Series

An MQTT historian is a subscriber that listens to your plant's MQTT topics and stores every value as time series, so operators can trend, compare and investigate what the namespace only shows in the moment. In this guide you will compare the main historian approaches, set up two working pipelines, Telegraf into InfluxDB and Python into TimescaleDB, and learn the design choices that decide whether your historian is trustworthy a year from now.

What an MQTT Historian Does

A classic process historian polls tags from SCADA or PLCs and compresses them into a proprietary archive. An MQTT historian inverts the flow: devices and gateways publish to a broker, and the historian is just one more subscriber. That has useful consequences:

• No extra load on the plant. The historian never polls a PLC; it reads what is already published.
• One integration point. Adding a machine to the namespace adds it to the historian automatically if the subscription covers its branch.
• Context from topics. With ISA-95 topics, site, area, line and machine become tags or columns without manual mapping.

What the historian must handle is everything the broker does not: long-term storage, efficient time-range queries, downsampling and retention. A broker keeps the latest retained value and short-lived session queues, not months of history. That division of labour is explained further in unified namespace vs data lake.

Comparing Historian Approaches

There is no single right answer. Many plants keep a hot historian with weeks or months of high-resolution data for operators and engineers, and push everything into a lake for long-term analytics.

Sample Data for Testing

Both pipelines expect topics like <prefix>/<site>/<area>/<line>/<machine>/telemetry with JSON payloads. If you have the gateway simulator from how to build a unified namespace, run it. Otherwise, this loop publishes simulated data to the public MQTT test broker. The public area is readable by everyone, so use your own random prefix and simulated values only:

PREFIX="connectmq/demo/h6n3"
while true; do
  TS=$(date -u +%Y-%m-%dT%H:%M:%SZ)
  TEMP=$(awk -v r=$RANDOM 'BEGIN{printf "%.1f", 38 + r % 60 / 10}')
  mosquitto_pub -h broker.connectmq.com -p 1883 -i hist-sim-h6n3 \
    -t "$PREFIX/dallas/packaging/line2/filler/telemetry" \
    -m "{\"ts\":\"$TS\",\"temp_c\":$TEMP,\"speed_bpm\":118}"
  sleep 5
done

Pipeline 1: Telegraf and InfluxDB

Telegraf's MQTT consumer plugin subscribes to the broker and writes to InfluxDB with no custom code. Start InfluxDB 2 in Docker with a development token:

docker run -d --name influxdb -p 8086:8086 \
  -e DOCKER_INFLUXDB_INIT_MODE=setup \
  -e DOCKER_INFLUXDB_INIT_USERNAME=admin \
  -e DOCKER_INFLUXDB_INIT_PASSWORD=change-me-please \
  -e DOCKER_INFLUXDB_INIT_ORG=plant \
  -e DOCKER_INFLUXDB_INIT_BUCKET=historian \
  -e DOCKER_INFLUXDB_INIT_ADMIN_TOKEN=dev-token-change-me \
  influxdb:2.7

Then create telegraf.conf. Topic parsing turns each ISA-95 level into an InfluxDB tag:

[agent]
  interval = "10s"
  flush_interval = "10s"

[[inputs.mqtt_consumer]]
  servers = ["tcp://broker.connectmq.com:1883"]
  topics = ["connectmq/demo/h6n3/+/+/+/+/telemetry"]
  qos = 1
  client_id = "telegraf-historian-h6n3"
  name_override = "telemetry"
  data_format = "json"
  json_time_key = "ts"
  json_time_format = "2006-01-02T15:04:05Z07:00"

  [[inputs.mqtt_consumer.topic_parsing]]
    topic = "connectmq/demo/h6n3/+/+/+/+/telemetry"
    tags = "_/_/_/site/area/line/machine/_"

[[outputs.influxdb_v2]]
  urls = ["http://localhost:8086"]
  token = "dev-token-change-me"
  organization = "plant"
  bucket = "historian"

Run it with telegraf --config telegraf.conf. Open the InfluxDB UI on port 8086, browse the historian bucket, and you will see a telemetry measurement with site, area, line and machine tags and numeric fields such as temp_c. The JSON parser keeps numeric fields by default; list string fields explicitly with json_string_fields if you need them. The Telegraf MQTT consumer documentation lists every option.

For a private ConnectMQ broker, change the server to ssl://YOUR_BROKER_HOST:8883, add username = "YOUR_USERNAME", password = "YOUR_PASSWORD" and set client_id to YOUR_CLIENT_ID, since client IDs are bound to the credentials issued in the portal. Drop the demo prefix and its parsing levels when your topics start with your enterprise name.

Pipeline 2: Python and TimescaleDB

When you want SQL and joins with production orders or batch records, TimescaleDB is a natural choice. Start it in Docker:

docker run -d --name timescaledb -p 5432:5432 -e POSTGRES_PASSWORD=historian timescale/timescaledb:latest-pg16

The subscriber below stores data in long format, one row per metric, which copes well with machines that publish different fields. Install the dependencies with pip install "paho-mqtt>=2.0" "psycopg[binary]".

import json
import os
from datetime import datetime, timezone

import paho.mqtt.client as mqtt
import psycopg

DSN = os.getenv("PG_DSN", "postgresql://postgres:historian@localhost:5432/postgres")
PREFIX = os.getenv("UNS_PREFIX", "connectmq/demo/h6n3")

conn = psycopg.connect(DSN, autocommit=True)
conn.execute("""
    CREATE TABLE IF NOT EXISTS telemetry (
        time    TIMESTAMPTZ NOT NULL,
        site    TEXT, area TEXT, line TEXT, machine TEXT,
        metric  TEXT NOT NULL,
        value   DOUBLE PRECISION
    )""")
conn.execute("SELECT create_hypertable('telemetry', 'time', if_not_exists => TRUE)")

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

def on_message(client, userdata, msg):
    if msg.retain:
        return  # retained values were already stored when first published
    site, area, line, machine, _ = msg.topic[len(PREFIX) + 1:].split("/")
    try:
        data = json.loads(msg.payload)
    except ValueError:
        return
    ts = data.pop("ts", None)
    when = datetime.fromisoformat(ts.replace("Z", "+00:00")) if ts else datetime.now(timezone.utc)
    rows = [(when, site, area, line, machine, k, float(v))
            for k, v in data.items() if isinstance(v, (int, float)) and not isinstance(v, bool)]
    with conn.cursor() as cur:
        cur.executemany(
            "INSERT INTO telemetry (time, site, area, line, machine, metric, value) "
            "VALUES (%s, %s, %s, %s, %s, %s, %s)", rows)

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="pg-historian-h6n3")
client.on_connect = on_connect
client.on_message = on_message
client.connect(os.getenv("MQTT_HOST", "broker.connectmq.com"), 1883, keepalive=60)
client.loop_forever()

Query five-minute averages per machine with standard SQL plus Timescale's time_bucket:

SELECT time_bucket('5 minutes', time) AS bucket, line, machine, avg(value) AS avg_temp
FROM telemetry
WHERE metric = 'temp_c' AND time > now() - interval '1 day'
GROUP BY bucket, line, machine
ORDER BY bucket;

Deciding What to Store

Storing every published value forever sounds safe, but it makes queries slower and costs grow quietly. Agree on a storage policy with operations and engineering before the historian goes live:

• Process values such as temperatures, pressures and speeds: store raw at the published rate for a few weeks, then keep aggregates such as one-minute minimum, maximum and average.
• Counters such as good and reject counts: store every change, because differences between readings are what matter for OEE and throughput.
• States and alarms: store every transition with its timestamp. They are low in volume and essential for downtime analysis.
• Events such as batch completions: store all of them, ideally in a table with their own fields rather than as generic metrics.
• Diagnostics such as gateway uptime or signal strength: keep a shorter retention unless you are investigating a specific problem.

Writing this down per namespace in your topic standard means the historian configuration follows the namespace design instead of growing ad hoc.

Design Decisions That Matter

1. Source timestamps over arrival time. Store the timestamp from the payload so buffered or delayed data lands at the right moment. Keep arrival time too if you need to measure latency.
2. Persistent sessions. Give the historian a fixed client ID and a persistent session (clean_session=False or an MQTT 5 session expiry) so the broker queues QoS 1 messages during a short restart.
3. Edge buffering for long outages. Broker queues cover brief gaps. For hours-long WAN outages, buffer at the gateway; see store and forward with MQTT.
4. Retained message handling. On subscribe, the broker replays retained values. Skip them or deduplicate on time and metric so restarts do not create duplicate rows.
5. Deadbands and report by exception. Publishing only meaningful changes reduces storage dramatically without losing information.
6. Retention and downsampling. Keep raw data for a defined period and continuous aggregates or downsampled buckets for longer.
7. Scale-out. With MQTT 5 shared subscriptions, several historian workers can share $share/historian/... so each message is written once. On ConnectMQ, shared subscriptions are available from the paid Serverless plan up.

ConnectMQ does not include a built-in database or historian connector, and it does not need one for this pattern: any of the subscribers above connects as a normal MQTT client. For more on modelling the data itself, read industrial time series data, and for the broader architecture see the industrial IoT MQTT guide.

Feed Your MQTT Historian From ConnectMQ

A historian is only as complete as the stream feeding it. ConnectMQ is a fully managed MQTT service with MQTT 3.1.1 and 5.0, QoS 1 and 2, persistent sessions, TLS and an ISA-95 aware portal, so your Telegraf or Python historian has a dependable source. Create a broker on the free Serverless plan at app.connectmq.com, no credit card required, or learn more about running a cloud MQTT broker for your plant.

Frequently Asked Questions

Is an MQTT broker a historian?

No. A broker routes messages and keeps the latest retained value per topic. A historian subscribes to the broker and stores the full time series for querying.

Should I choose InfluxDB or TimescaleDB?

Choose InfluxDB with Telegraf for the fastest configuration-only setup and Grafana dashboards. Choose TimescaleDB when you want SQL and joins with relational data such as orders and batches.

Can a commercial historian read from MQTT?

Many commercial historians offer MQTT or Sparkplug connectors. Check whether the connector supports your payload format and topic structure, and whether it can handle Sparkplug births and aliases if you use them.

How do I avoid losing data when the historian restarts?

Use a fixed client ID with a persistent session and QoS 1 so the broker queues messages briefly, and buffer at the edge for longer outages.