Building an IoT Data Pipeline: From MQTT to Time-Series Database and Dashboards

From Raw Sensor Readings to Something Useful

Getting telemetry off a device is only step one. This walks through the rest of the pipeline: ingesting MQTT messages, storing them efficiently in a time-series database, and building a dashboard that stays fast as data volume grows.

Why Time-Series Databases for IoT Data

IoT telemetry is high-volume, append-only, and almost always queried by time range. General-purpose relational databases handle this poorly at scale — a time-series database like InfluxDB or TimescaleDB is built specifically for this access pattern.

Bridging MQTT to InfluxDB

import paho.mqtt.client as mqtt
from influxdb_client import InfluxDBClient, Point
import json

influx = InfluxDBClient(url="http://localhost:8086", token=INFLUX_TOKEN, org="myorg")
write_api = influx.write_api()

def on_message(client, userdata, msg):
    data = json.loads(msg.payload)
    point = (
        Point("telemetry")
        .tag("device_id", data["device_id"])
        .field("temperature", data["temperature"])
        .field("humidity", data["humidity"])
    )
    write_api.write(bucket="iot-data", record=point)

mqtt_client = mqtt.Client()
mqtt_client.on_message = on_message
mqtt_client.connect("broker.example.com", 1883)
mqtt_client.subscribe("devices/+/telemetry")
mqtt_client.loop_forever()

Downsampling for Long-Term Storage

Raw per-second telemetry becomes expensive to store and slow to query over long time ranges. Continuous aggregation queries downsample old data automatically:

-- TimescaleDB continuous aggregate: hourly averages
CREATE MATERIALIZED VIEW telemetry_hourly
WITH (timescaledb.continuous) AS
SELECT device_id,
       time_bucket('1 hour', time) AS bucket,
       AVG(temperature) AS avg_temp,
       MAX(temperature) AS max_temp
FROM telemetry
GROUP BY device_id, bucket;

Querying for a Dashboard

SELECT bucket, avg_temp
FROM telemetry_hourly
WHERE device_id = 'sensor-01'
  AND bucket > NOW() - INTERVAL '7 days'
ORDER BY bucket;

Alerting on Anomalies

def check_anomaly(device_id, current_temp):
    baseline = get_recent_average(device_id, hours=24)
    if abs(current_temp - baseline) > 10:
        trigger_alert(device_id, current_temp, baseline)

Simple threshold-based alerting handles most cases; for genuinely noisy sensor data, a rolling standard deviation check reduces false positives compared to a fixed threshold.

Handling Data Volume at Scale

• Batch writes instead of writing on every single message — buffer for a few seconds and write in bulk
• Set retention policies so raw high-resolution data ages out automatically after downsampling
• Partition data by device or region if a single time-series instance becomes a bottleneck

Conclusion

An IoT pipeline’s long-term viability depends more on the storage and downsampling strategy than on the ingestion layer — MQTT reliably gets data in, but a time-series database with sensible retention and aggregation is what keeps dashboards fast as months of data accumulate.