Armazene mensagens MQTT no PostgreSQL e TimescaleDB com Python

Para armazenar mensagens MQTT no PostgreSQL, rode um pequeno consumidor que assina seus topics, interpreta cada payload e insere linhas em lotes, não uma transação por mensagem. Coloque o TimescaleDB por cima e a mesma tabela vira uma hypertable com particionamento por tempo, compressão e políticas de retenção.

Este tutorial constrói esse consumidor em Python com paho-mqtt 2.x e psycopg 3: um schema que funciona no Postgres puro e no Timescale, uma thread de gravação que agrupa por tamanho e por tempo, uma sessão persistente para reinícios e as consultas que provam que os dados chegaram.

Arquitetura

devices ──► MQTT broker ──► Python consumer ──► PostgreSQL / TimescaleDB
                            (on_message → queue → batch writer)

O callback MQTT precisa retornar rápido, porque o paho o executa na thread de rede. Então o on_message só interpreta o payload e coloca uma linha numa fila; uma thread de gravação separada esvazia a fila e grava em lotes. Essa separação é o design inteiro:

Se você é novo na biblioteca cliente, comece pelo tutorial do paho-mqtt 2.x.

Schema para Postgres e TimescaleDB

Suba um TimescaleDB local (é PostgreSQL normal com a extensão pré-instalada):

docker run -d --name tsdb -p 5432:5432 \
  -e POSTGRES_USER=mqtt -e POSTGRES_PASSWORD=mqtt -e POSTGRES_DB=telemetry \
  timescale/timescaledb:latest-pg16

Guarde algumas colunas tipadas que você consulta com frequência, mais o payload completo como jsonb, para que campos novos nunca quebrem a ingestão:

CREATE TABLE mqtt_messages (
  time       timestamptz NOT NULL,
  topic      text        NOT NULL,
  device_id  text        NOT NULL,
  payload    jsonb       NOT NULL
);

-- Somente TimescaleDB (sintaxe 2.13+); pule no PostgreSQL puro
SELECT create_hypertable('mqtt_messages', by_range('time'));

CREATE INDEX ON mqtt_messages (device_id, time DESC);

No PostgreSQL puro, o índice em (device_id, time DESC) basta para "últimos valores por dispositivo" e consultas por intervalo em volume moderado; adicione particionamento nativo por mês quando a tabela crescer. Em versões antigas do Timescale use create_hypertable('mqtt_messages', 'time').

O gravador em lote

Instale as dependências com pip install "paho-mqtt>=2" "psycopg[binary]>=3.1" e rode:

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)))   # bloqueia quando cheia = 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:      # banco fora: manter o lote e tentar de novo
            print(f"database unavailable, retrying: {e}")
            conn = None
            time.sleep(2)
        except psycopg.Error as e:                 # dados inválidos: descartar este lote
            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          # o broker guarda a sessão por 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()        # primeiro parar de receber...
client.loop_stop()
stop.set()                 # ...depois deixar o gravador esvaziar a fila
worker.join()

Alguns detalhes que valem a pena entender:

• executemany no psycopg 3 envia o lote de forma eficiente em modo pipeline, muito mais rápido que um execute mais commit por mensagem. Para taxas muito altas, troque por cur.copy("COPY mqtt_messages (time, topic, device_id, payload) FROM STDIN") e write_row().
• Jsonb(data) adapta o dict interpretado para jsonb. Guardar o payload bruto permite adicionar colunas tipadas depois e preenchê-las com SQL.
• A fila limitada é sua válvula de segurança: se o Postgres ficar lento, rows.put() bloqueia a thread MQTT, o cliente para de ler o socket e o broker segura as mensagens em vez de o seu processo ficar sem memória.
• A ordem de desligamento importa: desconecte do MQTT, depois esvazie e faça flush, para que nada recebido fique sem gravar.

Reinícios e garantias de entrega

Com clean_start=False, um client ID fixo e uma expiração de sessão, o broker enfileira mensagens QoS 1 enquanto o consumidor está fora e as entrega na reconexão (session present: True no log). Leia mais em sessões persistentes. Endpoints públicos de teste compartilhados, incluindo broker.connectmq.com, não guardam sessões offline, então teste esse comportamento no seu próprio broker privado.

Resta uma lacuna: o paho confirma uma mensagem QoS 1 assim que on_message retorna, antes de a linha ser commitada. Se o processo cair, as linhas que ainda estão na fila se perdem. Quando isso importa, crie o cliente com manual_ack=True, leve msg.mid e msg.qos junto com cada linha e chame client.ack(mid, qos) para cada mensagem, na ordem de recebimento, depois de conn.commit(). Lembre que o broker só envia um número limitado de mensagens não confirmadas por vez (Receive Maximum ou o limite inflight do broker), então mantenha FLUSH_SECS curto ou o throughput ficará limitado por essa janela.

Entrega at-least-once significa que duplicatas são possíveis após reconexões. Se elas importam, publique um número de sequência por dispositivo, adicione um índice único como (device_id, time, (payload->>'seq')) e insira com ON CONFLICT DO NOTHING. Em uma hypertable, índices únicos precisam incluir a coluna de tempo. Veja mensagens duplicadas para entender as causas.

Teste o pipeline

1. Suba o banco e aplique schema.sql com psql postgresql://mqtt:mqtt@localhost:5432/telemetry -f schema.sql.
2. Rode python mqtt_to_pg.py.
3. Publique dados de teste, pelo cliente MQTT online ou com um loop no shell:

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

1. Confira as linhas:

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 é uma função do TimescaleDB; no PostgreSQL puro use date_trunc('minute', time). Para testar reinícios, pare o consumidor, publique mais dez mensagens, inicie-o de novo e confirme que a contagem sobe em dez (em um broker que guarda sessões).

Publique leituras de teste pelo navegador

Retenção e escala

• Compressão e retenção (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');

• Mais consumidores: rode vários gravadores com uma assinatura compartilhada como $share/pg-writers/testmqtt/pg/+/telemetry, cada um com seu próprio client ID; o broker distribui as mensagens entre eles. Detalhes em assinaturas compartilhadas.
• Dashboards: o data source PostgreSQL do Grafana lê essa tabela diretamente; combine-o com o MQTT data source do Grafana para ter um painel ao vivo no topo.

Um bom pipeline de MQTT para SQL é propositalmente sem graça: interpreta rápido, enfileira com limite, grava em lotes e deixa a sessão persistente do broker cobrir os reinícios. Teste com algumas mensagens publicadas e depois com um reinício, e você saberá exatamente o que acontece com seus dados num dia ruim.

Perguntas frequentes