To store MQTT messages in PostgreSQL, run a small consumer that subscribes to your topics, parses each payload and inserts rows in batches — not one transaction per message. Add TimescaleDB on top and the same table becomes a hypertable with time partitioning, compression and retention policies.
This tutorial builds that consumer in Python with paho-mqtt 2.x and psycopg 3: a schema that works for plain Postgres and Timescale, a writer thread that batches by size and time, a persistent session for restarts, and the queries that prove the data landed.
Architecture
devices ──► MQTT broker ──► Python consumer ──► PostgreSQL / TimescaleDB
(on_message → queue → batch writer)The MQTT callback must return quickly, because paho runs it on the network thread. So on_message only parses the payload and puts a row on a queue; a separate writer thread drains the queue and writes batches. That split is the whole design:
| Part | Job | Why |
|---|---|---|
on_message | Parse JSON, build a row tuple | Keeps the network loop responsive (keep alive, acks) |
| Bounded queue | Buffer between MQTT and DB | Absorbs bursts; when full it slows MQTT down instead of growing forever |
| Writer thread | Flush every 500 rows or 1 s | One round trip and one commit per batch |
| Persistent session | Broker keeps messages while the consumer restarts | Restarts and deploys do not lose QoS 1 data |
If you are new to the client library, start with the paho-mqtt 2.x tutorial.
Schema for Postgres and TimescaleDB
Start a local TimescaleDB (it is plain PostgreSQL with the extension preinstalled):
docker run -d --name tsdb -p 5432:5432 \
-e POSTGRES_USER=mqtt -e POSTGRES_PASSWORD=mqtt -e POSTGRES_DB=telemetry \
timescale/timescaledb:latest-pg16Store a few typed columns you query often, plus the full payload as jsonb so new fields never break ingestion:
CREATE TABLE mqtt_messages (
time timestamptz NOT NULL,
topic text NOT NULL,
device_id text NOT NULL,
payload jsonb NOT NULL
);
-- TimescaleDB only (2.13+ syntax); skip on plain PostgreSQL
SELECT create_hypertable('mqtt_messages', by_range('time'));
CREATE INDEX ON mqtt_messages (device_id, time DESC);On plain PostgreSQL, the index on (device_id, time DESC) is enough for "latest values per device" and range queries at moderate volume; add native partitioning by month once the table gets large. On older Timescale versions use create_hypertable('mqtt_messages', 'time').
The batch writer
Install the dependencies with pip install "paho-mqtt>=2" "psycopg[binary]>=3.1" and run:
import json, os, queue, threading, time
from datetime import datetime, timezone
import paho.mqtt.client as mqtt
from paho.mqtt.packettypes import PacketTypes
from paho.mqtt.properties import Properties
import psycopg
from psycopg.types.json import Jsonb
BROKER = os.environ.get("MQTT_HOST", "broker.connectmq.com")
TOPIC = "testmqtt/pg/+/telemetry"
DSN = os.environ.get("PG_DSN", "postgresql://mqtt:mqtt@localhost:5432/telemetry")
BATCH_SIZE, FLUSH_SECS = 500, 1.0
INSERT = "INSERT INTO mqtt_messages (time, topic, device_id, payload) VALUES (%s, %s, %s, %s)"
rows: queue.Queue = queue.Queue(maxsize=50_000)
stop = threading.Event()
def on_connect(client, userdata, flags, reason_code, properties):
if reason_code.is_failure:
print(f"connect failed: {reason_code}")
return
print(f"connected, session present: {flags.session_present}")
client.subscribe(TOPIC, qos=1)
def on_message(client, userdata, msg):
try:
data = json.loads(msg.payload)
except (UnicodeDecodeError, json.JSONDecodeError):
print(f"skipping non-JSON payload on {msg.topic}")
return
ts = data.get("ts") if isinstance(data, dict) else None
when = (datetime.fromtimestamp(ts / 1000, timezone.utc)
if isinstance(ts, (int, float)) else datetime.now(timezone.utc))
device_id = msg.topic.split("/")[2]
rows.put((when, msg.topic, device_id, Jsonb(data))) # blocks when full = backpressure
def writer():
conn, batch = None, []
while not (stop.is_set() and rows.empty() and not batch):
deadline = time.monotonic() + FLUSH_SECS
while len(batch) < BATCH_SIZE and (left := deadline - time.monotonic()) > 0:
try:
batch.append(rows.get(timeout=left))
except queue.Empty:
break
if not batch:
continue
try:
if conn is None or conn.closed:
conn = psycopg.connect(DSN)
with conn.cursor() as cur:
cur.executemany(INSERT, batch)
conn.commit()
print(f"wrote {len(batch)} rows")
batch.clear()
except psycopg.OperationalError as e: # DB down: keep the batch, retry
print(f"database unavailable, retrying: {e}")
conn = None
time.sleep(2)
except psycopg.Error as e: # bad data: drop this batch
print(f"insert failed, dropping {len(batch)} rows: {e}")
conn.rollback()
batch.clear()
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
client_id="pg-writer-01", protocol=mqtt.MQTTv5)
client.username_pw_set(os.environ["MQTT_USER"], os.environ["MQTT_PASS"])
client.tls_set()
client.on_connect = on_connect
client.on_message = on_message
props = Properties(PacketTypes.CONNECT)
props.SessionExpiryInterval = 3600 # broker keeps our session for 1 h offline
client.connect(BROKER, 8883, clean_start=False, properties=props)
worker = threading.Thread(target=writer)
worker.start()
client.loop_start()
try:
while not stop.wait(1):
pass
except KeyboardInterrupt:
pass
client.disconnect() # stop receiving first...
client.loop_stop()
stop.set() # ...then let the writer drain the queue
worker.join()A few details worth understanding:
executemanyin psycopg 3 sends the batch efficiently in pipeline mode — far faster than oneexecutepluscommitper message. For very high rates, switch tocur.copy("COPY mqtt_messages (time, topic, device_id, payload) FROM STDIN")andwrite_row().Jsonb(data)adapts the parsed dict tojsonb. Keeping the raw payload means you can add typed columns later and backfill them with SQL.- The bounded queue is your safety valve: if Postgres slows down,
rows.put()blocks the MQTT thread, the client stops reading the socket, and the broker holds messages instead of your process running out of memory. - Shutdown order matters: disconnect from MQTT, then drain and flush, so nothing received is left unwritten.
Restarts and delivery guarantees
With clean_start=False, a fixed client ID and a session expiry, the broker queues QoS 1 messages while the consumer is down and delivers them on reconnect (session present: True in the log). Read more in persistent sessions. Shared public test endpoints, including broker.connectmq.com, do not keep offline sessions, so test this behavior on your own private broker.
There is one gap: paho acknowledges a QoS 1 message as soon as on_message returns, before the row is committed. If the process crashes, rows still in the queue are lost. When that matters, create the client with manual_ack=True, carry msg.mid and msg.qos along with each row, and call client.ack(mid, qos) for every message — in the order received — after conn.commit(). Remember that the broker only sends a limited number of unacknowledged messages at once (Receive Maximum or the broker's inflight limit), so keep FLUSH_SECS short or throughput will be capped by that window.
At-least-once delivery means duplicates are possible after reconnects. If they matter, publish a per-device sequence number, add a unique index such as (device_id, time, (payload->>'seq')), and insert with ON CONFLICT DO NOTHING. On a hypertable, unique indexes must include the time column. See duplicate messages for the causes.
Test the pipeline
- Start the database and apply
schema.sqlwithpsql postgresql://mqtt:mqtt@localhost:5432/telemetry -f schema.sql. - Run
python mqtt_to_pg.py. - Publish test data — from the online MQTT client or a shell loop:
for i in $(seq 1 20); do
mosquitto_pub -h broker.connectmq.com -p 8883 --capath /etc/ssl/certs \
-u "$MQTT_USER" -P "$MQTT_PASS" -q 1 -t testmqtt/pg/sensor-0$((i % 3))/telemetry \
-m "{\"ts\":$(date +%s000),\"seq\":$i,\"temperature\":$((20 + i % 5)).5}"
done- Check the rows:
SELECT device_id, count(*), max(time) FROM mqtt_messages GROUP BY device_id;
SELECT time_bucket('1 minute', time) AS minute, device_id,
avg((payload->>'temperature')::float) AS avg_temp
FROM mqtt_messages
WHERE time > now() - interval '1 hour'
GROUP BY 1, 2 ORDER BY 1;time_bucket is a TimescaleDB function; on plain PostgreSQL use date_trunc('minute', time). To test restarts, stop the consumer, publish ten more messages, start it again and confirm the count increases by ten (on a broker that keeps sessions).
Publish test readings from your browser
Retention and scaling
- Compression and retention (TimescaleDB):
ALTER TABLE mqtt_messages SET (timescaledb.compress, timescaledb.compress_segmentby = 'device_id');
SELECT add_compression_policy('mqtt_messages', INTERVAL '7 days');
SELECT add_retention_policy('mqtt_messages', INTERVAL '90 days');- More consumers: run several writers with a shared subscription such as
$share/pg-writers/testmqtt/pg/+/telemetry, each with its own client ID; the broker load-balances messages between them. Details in shared subscriptions. - Dashboards: Grafana's PostgreSQL data source reads this table directly; combine it with the Grafana MQTT data source for a live panel on top.
A good MQTT-to-SQL pipeline is boring by design: parse fast, queue with a limit, write in batches, and let the broker's persistent session cover restarts. Test it with a few published messages, then with a restart, and you will know exactly what happens to your data on a bad day.
Frequently asked questions
Should I insert every MQTT message into PostgreSQL individually?
Only at very low rates. One INSERT and one commit per message adds a network round trip and a transaction for every reading. Collect messages in memory and write them in batches every few hundred rows or every second, whichever comes first.
Do I need TimescaleDB or is plain PostgreSQL enough?
Plain PostgreSQL with a time index works well for moderate volumes. TimescaleDB adds automatic time partitioning, compression, retention policies and time_bucket queries, which matter once tables grow to many millions of rows.
How do I avoid losing messages if the database is down?
Subscribe with QoS 1 and a persistent session so the broker queues messages while your consumer is offline, and keep a bounded in-memory queue so a slow database applies backpressure. For strict at-least-once, acknowledge messages only after the database commit.
