MQTT para TimescaleDB: Armazene Séries Temporais IoT

Enviar dados MQTT para o TimescaleDB dá a você a confiabilidade e o SQL do PostgreSQL com armazenamento e consultas feitos para séries temporais. Neste tutorial você vai construir um assinante Python que grava em lote as mensagens de dispositivos vindas de um broker na nuvem em uma hypertable do TimescaleDB e, em seguida, acrescentar agregados contínuos, compressão e retenção para que os dashboards continuem rápidos à medida que os dados crescem.

Arquitetura

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

O broker roteia mensagens; ele não grava em bancos de dados. A ponte (bridge) é um cliente MQTT comum que assina, faz o parse e insere. Essa separação é um ponto forte: você pode reiniciar, escalar ou substituir a ponte sem mexer nos dispositivos, e o mesmo padrão funciona com qualquer broker MQTT gerenciado. O ConnectMQ cuida do lado MQTT; o TimescaleDB e a ponte rodam onde quer que você hospede os seus dados.

Componente	Papel	Neste tutorial
Dispositivos	Publicam telemetria em JSON	Simulados com `mosquitto_pub`
Broker	Roteia mensagens, mantém sessões	broker.connectmq.com, depois um broker privado
Ponte	Assina, valida, agrupa em lotes, insere	Python, paho-mqtt 2.x, psycopg 3
Banco de dados	Armazena e agrega séries temporais	TimescaleDB sobre PostgreSQL 16

Passo 1: rode o TimescaleDB

O início mais rápido é a imagem Docker oficial:

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

Conecte-se com psql -h localhost -U postgres e crie o schema.

Passo 2: projete a hypertable

Dispositivos IoT raramente publicam o mesmo conjunto de campos. Uma tabela estreita, com uma linha por valor de métrica, absorve novos sensores sem mudanças de schema e se comprime muito bem.

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 transforma a tabela em um conjunto de chunks particionados por tempo, um por dia neste caso. Consultas que filtram por tempo tocam apenas os chunks relevantes, e chunks antigos podem ser comprimidos ou descartados em uma única operação barata. Escolha um intervalo de chunk que faça os chunks mais recentes caberem com folga na memória; um dia serve para a maioria das frotas pequenas e médias.

Se todos os dispositivos enviam exatamente os mesmos campos, uma tabela larga (time, device_id, temperature, humidity, ...) também funciona e é um pouco mais simples de consultar. A ponte abaixo é fácil de adaptar para qualquer um dos casos.

Passo 3: a ponte de MQTT para TimescaleDB

Instale as dependências:

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

Os dispositivos publicam JSON como este em <prefix>/<device_id>/telemetry:

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

A ponte tem duas threads. A thread de rede MQTT apenas faz o parse e enfileira as linhas, então nunca fica bloqueada pelo banco de dados. Uma thread gravadora esvazia a fila e grava um lote com o COPY do PostgreSQL a cada segundo ou a cada 1.000 linhas, o que acontecer primeiro.

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):
    """Transforma uma mensagem MQTT em linhas (time, device_id, metric, value)."""
    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)  # mantém o lote e tenta de novo
    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)

    # Sessão persistente: o broker enfileira mensagens QoS 1 enquanto a ponte reinicia
    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()

Por que ela é construída assim

COPY em vez de INSERT linha a linha. O COPY em lote é a forma mais rápida de carregar muitas linhas no PostgreSQL e mantém baixo o número de transações.
Uma fila limitada. Se o banco ficar fora do ar por muito tempo, a memória não cresce sem limite. A ponte registra em log e descarta em vez de travar; dimensione MAX_QUEUE para a queda que você quer suportar.
Sessão MQTT persistente. Com clean_start=False, um client ID fixo e uma expiração de sessão, o broker guarda as mensagens QoS 1 enfileiradas enquanto a ponte reinicia. No servidor MQTT público, as sessões duram até 30 minutos. Defina um MQTT_CLIENT_ID estável para que isso funcione; um ID aleatório cria uma sessão nova a cada vez. Contexto em sessões persistentes MQTT.
Timestamps do dispositivo quando disponíveis. Dispositivos que guardam leituras offline as enviam com atraso; usar o ts deles mantém a série temporal correta.

Passo 4: teste o pipeline

Inicie a ponte e publique algumas leituras no servidor MQTT público (gratuito), pensado para testes. A área pública pode ser lida por qualquer pessoa, então use dados sintéticos:

python mqtt_to_timescaledb.py

# em outro 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

Confira os dados:

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

Passo 5: agregados contínuos para dashboards rápidos

Dashboards que plotam um mês de dados brutos por segundo são lentos. Um agregado contínuo pré-calcula os resumos e os atualiza de forma incremental em segundo plano:

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 controla até quanto tempo para trás os dados atrasados são considerados em cada atualização. Se os dispositivos podem ficar offline por mais de duas horas e depois enviar as leituras guardadas, aumente esse valor. Consulte a view exatamente como uma tabela:

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

Passo 6: compressão e retenção

Dados de sensores se comprimem extremamente bem quando agrupados por dispositivo e métrica:

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');

Isso mantém 90 dias de dados brutos, comprime tudo o que tem mais de uma semana e deixa o agregado de um minuto vivo para tendências de longo prazo. As versões recentes do TimescaleDB também descrevem a compressão como "columnstore"; as políticas acima continuam funcionando.

Visualizando os dados no Grafana

A fonte de dados PostgreSQL nativa do Grafana funciona com o TimescaleDB sem alterações. Adicione uma fonte de dados apontando para o banco, ative a opção TimescaleDB nas configurações dela e conecte-se com um papel somente leitura, e não com o superusuário que a ponte usa. Escreva as consultas dos painéis contra o agregado contínuo e deixe as macros do Grafana inserirem o intervalo de tempo do dashboard:

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

Use a tabela bruta readings apenas para intervalos curtos em que o detalhe por segundo importa; qualquer coisa que abranja dias deve consultar o agregado. Se preferir dispensar o banco para uma visualização ao vivo rápida, o Grafana com uma fonte de dados MQTT pode plotar topics diretamente, mas não guarda histórico.

Checklist de produção

Saia do servidor MQTT público para um broker privado: crie um broker no portal do ConnectMQ e defina MQTT_USERNAME, MQTT_PASSWORD e MQTT_CLIENT_ID com os valores gerados.
Mantenha TLS na porta 8883.
Rode a ponte sob o systemd ou como contêiner com restart always.
Use QoS 1 de ponta a ponta; o plano gratuito suporta QoS até 1.
Torne as gravações idempotentes se duplicatas importarem: o QoS 1 pode reentregar. Um índice único em (device_id, metric, time) mais uma tabela de staging com ON CONFLICT DO NOTHING resolve isso.
Monitore a profundidade da fila e as linhas por segundo nos logs da ponte.
Para escalar horizontalmente, rode várias pontes com shared subscriptions do MQTT 5 ($share/tsdb/<prefix>/+/telemetry), disponíveis nos planos ConnectMQ Serverless pagos e superiores. Veja shared subscriptions no MQTT.

Solução de problemas

Sintoma	Causa provável
`relation "readings" does not exist`	Schema criado em um banco diferente daquele para o qual `PG_DSN` aponta
A ponte conecta, mas não grava nada	O topic não termina em `/telemetry` ou o prefixo diverge
As linhas têm o horário errado	Relógio do dispositivo errado; considere ignorar `ts` muito no futuro
Lacunas após reinícios	Client ID aleatório, portanto sem sessão persistente
O agregado está vazio	`WITH NO DATA` e a política ainda não rodou; execute `CALL refresh_continuous_aggregate('readings_1m', NULL, NULL);`

Se preferir começar com um armazenamento mais simples, compare esta abordagem com salvar mensagens MQTT no MongoDB ou registrar mensagens MQTT em CSV. Para mais padrões de clientes, o guia de bibliotecas cliente MQTT é uma boa próxima leitura.

Comece a usar o ConnectMQ

Aponte a sua ponte para TimescaleDB para um broker privado. Crie um broker Serverless gratuito em app.connectmq.com, sem cartão de crédito, defina as credenciais geradas como variáveis de ambiente e use o explorador de topics ao vivo do portal para confirmar o que os seus dispositivos publicam antes que chegue ao banco de dados.

Perguntas frequentes

O ConnectMQ consegue gravar dados MQTT diretamente no TimescaleDB?

Não, o ConnectMQ é o broker MQTT e não inclui integrações com bancos de dados. Rode um assinante, como a ponte Python deste tutorial, o Telegraf ou o Node-RED, para gravar os dados.

Devo usar uma tabela estreita ou larga para dados MQTT no TimescaleDB?

Use uma tabela estreita (time, device, metric, value) quando os dispositivos enviam campos variados, e uma tabela larga quando todos os dispositivos enviam o mesmo conjunto fixo. Tabelas estreitas não precisam de migrações quando você adiciona sensores.

Como evitar perder mensagens MQTT quando o banco de dados está fora do ar?

Faça buffer na ponte com uma fila limitada e repita as gravações, e use uma sessão MQTT persistente com QoS 1 para que o broker guarde as mensagens enquanto a ponte reinicia.

O PostgreSQL puro basta em vez do TimescaleDB?

Para volumes pequenos, sim. O TimescaleDB acrescenta particionamento automático, agregados contínuos, compressão e políticas de retenção que se tornam valiosos à medida que os dados crescem.

MQTT para TimescaleDB: Armazenando Séries Temporais IoT no PostgreSQL