Historian & Time-Series

Industrial MQTT → TimescaleDB pipeline, anomaly detection, downsampling

Overview

The Historian provides an industrial-grade MQTT-to-TimescaleDB time-series pipeline. It subscribes to MQTT topics on the Unified Namespace (UNS), automatically creates hypertables, and stores data with configurable retention policies, downsampling, and crash recovery.

Architecture

// Historian Pipeline

MQTT Broker (UNS Topics)

│

├── Topic Profiles (routing rules)

▼

Historian Service

│

├── Disk Buffer (crash recovery)

├── Backpressure control

├── COPY protocol (bulk insert)

▼

TimescaleDB (PostgreSQL)

│

├── Hypertables (auto-created)

├── Retention policies

└── Continuous aggregates (downsampling)

Features

Auto Table Creation — Hypertables are created automatically when new MQTT topics are first seen
Retention Policies — Configurable per-table data retention (e.g., 90 days raw, 1 year aggregated)
Downsampling — Continuous aggregates for 1-minute, 1-hour, and 1-day rollups
Disk Buffer — Writes to local disk when the database is unreachable, replays on reconnect
Backpressure — Throttles ingestion when the database falls behind to prevent memory exhaustion
COPY Protocol — Uses PostgreSQL COPY for bulk inserts, significantly faster than row-by-row INSERT
Dead-Letter Queue — Messages that fail 3 retries are moved to a dead-letter topic for investigation

MCP Tools (6 History Tools)

The Historian exposes 6 tools via MCP for LLM-driven time-series analysis:

history_get_trend

Retrieve time-series data for a variable over a time range with optional downsampling

history_compare

Compare trends of multiple variables side-by-side over the same time window

history_aggregate

Compute aggregations (min, max, avg, sum, count) over configurable time buckets

history_anomalies

Detect anomalies using statistical methods (z-score, IQR) on historical data

history_machines

List all machines with historian data, including variable counts and time ranges

history_variables

List all recorded variables for a machine with metadata (type, unit, sample count)

Topic Profiles

Topic profiles define how MQTT topics are routed to TimescaleDB tables. Each profile specifies:

Topic Pattern — MQTT topic filter (supports wildcards)
Target Table — TimescaleDB hypertable name
Field Mapping — JSON path to extract timestamp, value, and tags
Retention — Per-profile data retention period

// Example topic profile

{

topic: 'uns/bz-1/cnc/+/temperature',

table: 'bde_data',

fields: { ts: '$.timestamp', value: '$.value' },

retention: '90d'

}

REST API

The Historian exposes a REST API for management and data exploration:

Route Management

Create, update, and delete MQTT topic subscriptions and routing rules.

Profile Management

Configure topic profiles with field mappings, retention, and downsampling settings.

Data Explorer

Query historical data with time range, aggregation, and filtering parameters.

Resilience

The Historian is designed for industrial reliability:

Disk Buffer — When TimescaleDB is unreachable, messages are buffered to local disk and replayed automatically on reconnect
Backpressure — If the write queue exceeds a configurable threshold, the MQTT client pauses consumption to prevent OOM
3-Retry Dead-Letter — Failed messages are retried 3 times with exponential backoff before being moved to a dead-letter topic
COPY Protocol — Bulk inserts via PostgreSQL COPY are atomic and significantly faster than individual INSERTs

Tip

The Historian works hand-in-hand with the Knowledge Graph — machines and sensors discovered in the KG automatically get historian subscriptions.

← Knowledge Graph OPC-UA Server Integration →