MQTT to TimescaleDB: Store IoT Time Series Data

Sending MQTT to TimescaleDB gives you the reliability and SQL of PostgreSQL with storage and queries built for time series. In this tutorial you will build a Python subscriber that batches device messages from a cloud broker into a TimescaleDB hypertable, then add continuous aggregates, compression and retention so dashboards stay fast as data grows.

Architecture

devices --MQTT--> broker (ConnectMQ) --MQTT--> Python bridge --COPY--> TimescaleDB --SQL--> Grafana / apps

The broker routes messages; it does not write to databases. The bridge is an ordinary MQTT client that subscribes, parses and inserts. That separation is a strength: you can restart, scale or replace the bridge without touching devices, and the same pattern works with any managed MQTT broker. ConnectMQ handles the MQTT side; TimescaleDB and the bridge run wherever you host your data.

Component	Role	In this tutorial
Devices	Publish JSON telemetry	Simulated with `mosquitto_pub`
Broker	Route messages, hold sessions	broker.connectmq.com, then a private broker
Bridge	Subscribe, validate, batch, insert	Python, paho-mqtt 2.x, psycopg 3
Database	Store and aggregate time series	TimescaleDB on PostgreSQL 16

Step 1: Run TimescaleDB

The quickest start is the official Docker image:

docker run -d --name tsdb -p 5432:5432 \
  -e POSTGRES_PASSWORD=change-me \
  -v tsdb-data:/var/lib/postgresql/data \
  timescale/timescaledb:latest-pg16

Connect with psql -h localhost -U postgres and create the schema.

Step 2: Design the Hypertable

IoT devices rarely publish the same set of fields. A narrow table, one row per metric value, absorbs new sensors without schema changes and compresses very well.

CREATE EXTENSION IF NOT EXISTS timescaledb;

CREATE TABLE readings (
    time       TIMESTAMPTZ      NOT NULL,
    device_id  TEXT             NOT NULL,
    metric     TEXT             NOT NULL,
    value      DOUBLE PRECISION NOT NULL
);

SELECT create_hypertable('readings', by_range('time', INTERVAL '1 day'));

CREATE INDEX readings_device_metric_time
    ON readings (device_id, metric, time DESC);

create_hypertable turns the table into a set of time-partitioned chunks, one per day here. Queries that filter on time touch only the relevant chunks, and old chunks can be compressed or dropped in one cheap operation. Pick a chunk interval so the most recent chunks fit comfortably in memory; one day suits most small and medium fleets.

If every device sends exactly the same fields, a wide table (time, device_id, temperature, humidity, ...) is also fine and slightly simpler to query. The bridge below is easy to adapt either way.

Step 3: The MQTT to TimescaleDB Bridge

Install the dependencies:

pip install "paho-mqtt>=2,<3" "psycopg[binary]>=3.1"

Devices publish JSON like this to <prefix>/<device_id>/telemetry:

{"ts": 1792137600.5, "temperature": 23.4, "humidity": 41.0, "battery": 3.71}

The bridge has two threads. The MQTT network thread only parses and enqueues rows, so it never blocks on the database. A writer thread drains the queue and writes a batch with PostgreSQL COPY every second or every 1,000 rows, whichever comes first.

mqtt_to_timescaledb.py:

import json
import logging
import os
import queue
import signal
import threading
import time
import uuid
from datetime import datetime, timezone

import paho.mqtt.client as mqtt
import psycopg
from paho.mqtt.packettypes import PacketTypes
from paho.mqtt.properties import Properties

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("bridge")

MQTT_HOST = os.getenv("MQTT_HOST", "broker.connectmq.com")
MQTT_PORT = int(os.getenv("MQTT_PORT", "8883"))
MQTT_USER = os.getenv("MQTT_USERNAME")
MQTT_PASS = os.getenv("MQTT_PASSWORD")
CLIENT_ID = os.getenv("MQTT_CLIENT_ID") or f"tsdb-bridge-{uuid.uuid4().hex[:8]}"
PREFIX = os.getenv("MQTT_TOPIC_PREFIX", "connectmq/demo/tsdb-5b7e21")
DSN = os.getenv("PG_DSN", "postgresql://postgres:change-me@localhost:5432/postgres")

BATCH_SIZE = 1000
FLUSH_SECONDS = 1.0
MAX_QUEUE = 100_000

rows: "queue.Queue[tuple]" = queue.Queue(maxsize=MAX_QUEUE)
stop = threading.Event()


def parse(topic: str, payload: bytes):
    """Turn one MQTT message into (time, device_id, metric, value) rows."""
    parts = topic.split("/")
    if len(parts) < 2 or parts[-1] != "telemetry":
        return []
    device_id = parts[-2]
    data = json.loads(payload)
    if not isinstance(data, dict):
        return []
    ts = data.pop("ts", None)
    when = (datetime.fromtimestamp(float(ts), tz=timezone.utc)
            if isinstance(ts, (int, float)) else datetime.now(timezone.utc))
    out = []
    for metric, value in data.items():
        if isinstance(value, bool):
            value = float(value)
        if isinstance(value, (int, float)):
            out.append((when, device_id, metric[:64], float(value)))
    return out


def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        log.error("connect failed: %s", reason_code)
        return
    log.info("connected (session present: %s)", flags.session_present)
    client.subscribe(f"{PREFIX}/+/telemetry", qos=1)


def on_message(client, userdata, msg):
    try:
        for row in parse(msg.topic, msg.payload):
            rows.put_nowait(row)
    except (ValueError, json.JSONDecodeError):
        log.warning("bad payload on %s", msg.topic)
    except queue.Full:
        log.error("queue full, dropping message from %s", msg.topic)


def writer():
    conn = None
    batch = []
    last_flush = time.monotonic()
    while not (stop.is_set() and rows.empty() and not batch):
        try:
            batch.append(rows.get(timeout=0.2))
            if len(batch) < BATCH_SIZE and time.monotonic() - last_flush < FLUSH_SECONDS:
                continue
        except queue.Empty:
            if not batch:
                continue
        try:
            if conn is None or conn.closed:
                conn = psycopg.connect(DSN, autocommit=False)
            with conn.cursor() as cur:
                with cur.copy("COPY readings (time, device_id, metric, value) FROM STDIN") as cp:
                    for r in batch:
                        cp.write_row(r)
            conn.commit()
            log.info("wrote %d rows", len(batch))
            batch.clear()
            last_flush = time.monotonic()
        except psycopg.Error as e:
            log.error("database error, retrying in 5 s: %s", e)
            if conn is not None:
                conn.close()
            conn = None
            time.sleep(5)  # keep the batch and retry
    if conn is not None:
        conn.close()


def main():
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID,
                         protocol=mqtt.MQTTv5)
    if MQTT_USER:
        client.username_pw_set(MQTT_USER, MQTT_PASS)
    if MQTT_PORT == 8883:
        client.tls_set()
    client.on_connect = on_connect
    client.on_message = on_message
    client.reconnect_delay_set(min_delay=1, max_delay=60)

    # Persistent session: the broker queues QoS 1 messages while the bridge restarts
    props = Properties(PacketTypes.CONNECT)
    props.SessionExpiryInterval = 1800

    t = threading.Thread(target=writer, daemon=True)
    t.start()

    def shutdown(*_):
        log.info("shutting down")
        client.disconnect()
        stop.set()

    signal.signal(signal.SIGINT, shutdown)
    signal.signal(signal.SIGTERM, shutdown)

    client.connect(MQTT_HOST, MQTT_PORT, keepalive=30, clean_start=False, properties=props)
    client.loop_forever(retry_first_connection=True)
    t.join(timeout=30)


if __name__ == "__main__":
    main()

Why It Is Built This Way

COPY instead of row-by-row INSERT. Batched COPY is the fastest way to load many rows into PostgreSQL and keeps the number of transactions low.
A bounded queue. If the database is down for long, memory does not grow without limit. The bridge logs and drops instead of crashing; size MAX_QUEUE for the outage you want to survive.
Persistent MQTT session. With clean_start=False, a fixed client ID and a session expiry, the broker keeps queued QoS 1 messages while the bridge restarts. On the public broker sessions last up to 30 minutes. Set a stable MQTT_CLIENT_ID for this to work; a random ID creates a new session each time. Background in MQTT persistent sessions.
Device timestamps when available. Devices that buffer offline readings send them late; using their own ts keeps the time series correct.

Step 4: Test the Pipeline

Start the bridge, then publish a few readings to the public MQTT test broker. The public area is readable by anyone, so use synthetic data:

python mqtt_to_timescaledb.py

# in another terminal
for i in $(seq 1 20); do
  mosquitto_pub -h broker.connectmq.com -p 1883 -i tsdb-sim-01 -q 1 \
    -t "connectmq/demo/tsdb-5b7e21/sensor-01/telemetry" \
    -m "{\"temperature\": 2$((RANDOM % 10)).$((RANDOM % 10)), \"humidity\": 4$((RANDOM % 10))}"
  sleep 1
done

Check the data:

SELECT device_id, metric, count(*), min(time), max(time)
FROM readings GROUP BY device_id, metric;

Step 5: Continuous Aggregates for Fast Dashboards

Dashboards that plot a month of raw per-second data are slow. A continuous aggregate precomputes rollups and refreshes them incrementally in the background:

CREATE MATERIALIZED VIEW readings_1m
WITH (timescaledb.continuous) AS
SELECT time_bucket(INTERVAL '1 minute', time) AS bucket,
       device_id,
       metric,
       avg(value) AS avg_value,
       min(value) AS min_value,
       max(value) AS max_value,
       count(*)   AS samples
FROM readings
GROUP BY bucket, device_id, metric
WITH NO DATA;

SELECT add_continuous_aggregate_policy('readings_1m',
    start_offset      => INTERVAL '2 hours',
    end_offset        => INTERVAL '1 minute',
    schedule_interval => INTERVAL '1 minute');

start_offset controls how far back late data is picked up on each refresh. If devices can be offline for longer than two hours and then upload buffered readings, widen it. Query the view exactly like a table:

SELECT bucket, avg_value
FROM readings_1m
WHERE device_id = 'sensor-01' AND metric = 'temperature'
  AND bucket > now() - INTERVAL '24 hours'
ORDER BY bucket;

Step 6: Compression and Retention

Sensor data compresses extremely well when grouped by device and metric:

ALTER TABLE readings SET (
    timescaledb.compress,
    timescaledb.compress_segmentby = 'device_id, metric',
    timescaledb.compress_orderby   = 'time DESC'
);

SELECT add_compression_policy('readings', INTERVAL '7 days');
SELECT add_retention_policy('readings', INTERVAL '90 days');

This keeps 90 days of raw data, compresses everything older than a week, and lets the one-minute aggregate live on for long-term trends. Recent TimescaleDB versions also describe compression as a "columnstore"; the policies above continue to work.

Visualizing the Data in Grafana

Grafana's built-in PostgreSQL data source works with TimescaleDB unchanged. Add a data source pointing at the database, enable the TimescaleDB option in its settings, and connect with a read-only role rather than the superuser the bridge uses. Write panel queries against the continuous aggregate and let Grafana's macros insert the dashboard's time range:

SELECT bucket AS "time", device_id AS metric, avg_value
FROM readings_1m
WHERE $__timeFilter(bucket) AND metric = 'temperature'
ORDER BY bucket;

Use the raw readings table only for short ranges where per-second detail matters; anything spanning days should hit the aggregate. If you prefer to skip the database for a quick live view, Grafana with an MQTT data source can plot topics directly, but it keeps no history.

Production Checklist

Move from the public broker to a private one: create a broker in the ConnectMQ portal, then set MQTT_USERNAME, MQTT_PASSWORD and MQTT_CLIENT_ID to the generated values.
Keep TLS on port 8883.
Run the bridge under systemd or as a container with restart always.
Use QoS 1 end to end; the free plan supports QoS up to 1.
Make writes idempotent if duplicates matter: QoS 1 can redeliver. A unique index on (device_id, metric, time) plus a staging table with ON CONFLICT DO NOTHING handles it.
Monitor queue depth and rows per second in the bridge logs.
To scale out, run several bridges with MQTT 5 shared subscriptions ($share/tsdb/<prefix>/+/telemetry), available on paid ConnectMQ Serverless plans and above. See MQTT shared subscriptions.

Troubleshooting

Symptom	Likely cause
`relation "readings" does not exist`	Schema created in a different database than `PG_DSN` points to
Bridge connects but writes nothing	Topic does not end in `/telemetry` or prefix mismatch
Rows have the wrong time	Device clock wrong; consider ignoring `ts` that is far in the future
Gaps after restarts	Random client ID, so no persistent session
Aggregate is empty	`WITH NO DATA` and the policy has not run yet; call `CALL refresh_continuous_aggregate('readings_1m', NULL, NULL);`

If you would rather start with a simpler store, compare this approach with saving MQTT messages to MongoDB or logging MQTT messages to CSV. For more client patterns, the MQTT client libraries guide is a good next read.

Get Started with ConnectMQ

Point your TimescaleDB bridge at a private broker. Create a free Serverless broker at app.connectmq.com with no credit card, set the generated credentials as environment variables, and use the portal's live topic explorer to confirm what your devices publish before it reaches the database.

Frequently Asked Questions

Can ConnectMQ write MQTT data to TimescaleDB directly?

No, ConnectMQ is the MQTT broker and does not include database integrations. Run a subscriber such as the Python bridge in this tutorial, Telegraf or Node-RED to write the data.

Should I use a narrow or wide table for MQTT data in TimescaleDB?

Use a narrow table (time, device, metric, value) when devices send varying fields, and a wide table when every device sends the same fixed set. Narrow tables need no migrations when you add sensors.

How do I avoid losing MQTT messages when the database is down?

Buffer in the bridge with a bounded queue and retry writes, and use a persistent MQTT session with QoS 1 so the broker holds messages while the bridge restarts.

Is plain PostgreSQL enough instead of TimescaleDB?

For small volumes, yes. TimescaleDB adds automatic partitioning, continuous aggregates, compression and retention policies that become valuable as data grows.

MQTT to TimescaleDB: Storing IoT Time Series in PostgreSQL