Como Armazenar Dados MQTT no InfluxDB com Python

Para armazenar dados MQTT no InfluxDB, você precisa de um assinante que transforme cada mensagem em um ponto de série temporal com o measurement, as tags, os fields e o timestamp corretos. Neste tutorial você vai escrever esse assinante em Python com paho-mqtt e o cliente oficial do InfluxDB, derivar tags automaticamente dos níveis do topic, gravar em lotes e tratar os problemas de payload que dispositivos reais produzem.

Por que escrever uma ponte em Python

O Telegraf consegue assinar MQTT e gravar no InfluxDB apenas com configuração, e para muitas equipes essa é a escolha certa. Uma ponte curta em Python vale a pena quando você precisa de uma lógica que a configuração não expressa com conforto: validar valores contra limites físicos, converter unidades, descartar duplicatas, enriquecer pontos com metadados de uma tabela de consulta ou tratar vários formatos de payload nos mesmos topics. Ela também é fácil de ler, testar e estender.

A ponte é um cliente MQTT comum. Ela se conecta ao broker, assina os topics e grava no InfluxDB pela API HTTP. Nenhum plugin ou integração do lado do broker está envolvido, o que significa que o mesmo código funciona com qualquer broker compatível com o padrão, incluindo o servidor MQTT público (gratuito) em broker.connectmq.com e um broker privado da ConnectMQ.

Modelo de dados: de topics para tags

O InfluxDB organiza os dados em measurements, tags e fields. Tags são strings indexadas pelas quais você filtra e agrupa. Fields são os valores que você plota e agrega. Um topic MQTT bem projetado já contém as tags de que você precisa:

connectmq/demo/a1b2c3/line1/sensor-0042/telemetry
   |       |     |      |        |          |
 ignore ignore ignore  line    device   measurement

O payload fornece os fields e o timestamp:

{"device": "sensor-0042", "ts": 1760600000123, "temperature": 21.84, "humidity": 44.2}

O ponto resultante no InfluxDB fica assim em line protocol:

telemetry,line=line1,device=sensor-0042 temperature=21.84,humidity=44.2 1760600000123000000

Mapear os níveis do topic pela posição mantém a ponte genérica: se você adicionar uma nova linha ou dispositivo, nada no código muda.

Configure o InfluxDB

Se você ainda não tem o InfluxDB 2, rode-o no Docker. As variáveis de inicialização criam uma organização, um bucket e um token de administrador na primeira execução.

docker run -d --name influxdb -p 8086:8086 \
  -v influxdb-data:/var/lib/influxdb2 \
  -e DOCKER_INFLUXDB_INIT_MODE=setup \
  -e DOCKER_INFLUXDB_INIT_USERNAME=admin \
  -e DOCKER_INFLUXDB_INIT_PASSWORD=change-me-please \
  -e DOCKER_INFLUXDB_INIT_ORG=iot \
  -e DOCKER_INFLUXDB_INIT_BUCKET=sensors \
  -e DOCKER_INFLUXDB_INIT_RETENTION=30d \
  -e DOCKER_INFLUXDB_INIT_ADMIN_TOKEN=my-super-secret-token \
  influxdb:2.7

Para qualquer coisa além de um teste local, crie na interface do InfluxDB um token separado com permissão de escrita apenas no bucket sensors e use-o no lugar do token de administrador.

Instale os pacotes Python:

pip install "paho-mqtt>=2.0" influxdb-client

A ponte completa para armazenar dados MQTT no InfluxDB

Salve o código a seguir como mqtt_influx_bridge.py. A configuração vem de variáveis de ambiente, então o mesmo arquivo serve para testes e produção.

import json
import logging
import os
import signal
import time
import uuid

import paho.mqtt.client as mqtt
from influxdb_client import InfluxDBClient, Point, WritePrecision, WriteOptions

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

# --- configuração ----------------------------------------------------------
MQTT_HOST = os.getenv("MQTT_HOST", "broker.connectmq.com")
MQTT_PORT = int(os.getenv("MQTT_PORT", "1883"))
MQTT_TLS = os.getenv("MQTT_TLS", "0") == "1"
MQTT_USER = os.getenv("MQTT_USERNAME")
MQTT_PASS = os.getenv("MQTT_PASSWORD")
MQTT_CLIENT_ID = os.getenv("MQTT_CLIENT_ID", f"influx-bridge-{uuid.uuid4().hex[:8]}")
MQTT_TOPIC = os.getenv("MQTT_TOPIC", "connectmq/demo/a1b2c3/+/+/telemetry")
# Um nome por nível do topic: "_" ignora um nível, "@measurement" nomeia o measurement
TOPIC_SCHEMA = os.getenv("TOPIC_SCHEMA", "_/_/_/line/device/@measurement").split("/")

INFLUX_URL = os.getenv("INFLUX_URL", "http://localhost:8086")
INFLUX_TOKEN = os.environ["INFLUX_TOKEN"]
INFLUX_ORG = os.getenv("INFLUX_ORG", "iot")
INFLUX_BUCKET = os.getenv("INFLUX_BUCKET", "sensors")

SKIP_KEYS = {"ts", "device"}  # chaves do payload que não são fields

# --- InfluxDB --------------------------------------------------------------
influx = InfluxDBClient(url=INFLUX_URL, token=INFLUX_TOKEN, org=INFLUX_ORG)
write_api = influx.write_api(write_options=WriteOptions(
    batch_size=500, flush_interval=5_000, retry_interval=5_000, max_retries=5))

stats = {"written": 0, "skipped": 0}


def to_point(topic: str, payload: bytes):
    levels = topic.split("/")
    if len(levels) != len(TOPIC_SCHEMA):
        return None
    try:
        data = json.loads(payload)
    except (UnicodeDecodeError, json.JSONDecodeError):
        return None
    if not isinstance(data, dict):
        return None

    measurement = "mqtt"
    tags = {}
    for name, value in zip(TOPIC_SCHEMA, levels):
        if name == "_":
            continue
        if name == "@measurement":
            measurement = value
        else:
            tags[name] = value

    point = Point(measurement)
    for k, v in tags.items():
        point = point.tag(k, v)

    fields = 0
    for key, value in data.items():
        if key in SKIP_KEYS:
            continue
        if isinstance(value, bool):
            point = point.field(key, value)
        elif isinstance(value, (int, float)):
            point = point.field(key, float(value))  # evita conflitos de tipo int/float
        elif isinstance(value, str) and len(value) <= 64:
            point = point.field(key, value)
        else:
            continue
        fields += 1
    if fields == 0:
        return None

    ts = data.get("ts")
    if isinstance(ts, (int, float)) and ts > 1e12:  # parece estar em milissegundos
        point = point.time(int(ts), WritePrecision.MS)
    else:
        point = point.time(time.time_ns(), WritePrecision.NS)
    return point


def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        log.error("MQTT connect failed: %s", reason_code)
        return
    client.subscribe(MQTT_TOPIC, qos=1)
    log.info("subscribed to %s", MQTT_TOPIC)


def on_message(client, userdata, msg):
    point = to_point(msg.topic, msg.payload)
    if point is None:
        stats["skipped"] += 1
        return
    write_api.write(bucket=INFLUX_BUCKET, record=point)  # com buffer, não bloqueante
    stats["written"] += 1


def main():
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=MQTT_CLIENT_ID)
    if MQTT_USER:
        client.username_pw_set(MQTT_USER, MQTT_PASS)
    if MQTT_TLS:
        client.tls_set()
    client.on_connect = on_connect
    client.on_message = on_message
    client.reconnect_delay_set(min_delay=1, max_delay=60)
    client.connect_async(MQTT_HOST, MQTT_PORT, keepalive=30)
    client.loop_start()

    running = True

    def stop(*_):
        nonlocal running
        running = False

    signal.signal(signal.SIGINT, stop)
    signal.signal(signal.SIGTERM, stop)

    while running:
        time.sleep(30)
        log.info("written=%d skipped=%d", stats["written"], stats["skipped"])

    client.loop_stop()
    client.disconnect()
    write_api.close()  # descarrega os pontos pendentes
    influx.close()


if __name__ == "__main__":
    main()

Execute a ponte e inicie uma fonte de dados. O simulador de dispositivos MQTT publica exatamente neste formato; defina MQTT_TOPIC com o prefixo dele:

export INFLUX_TOKEN=my-super-secret-token
export MQTT_TOPIC="connectmq/demo/3f9a1c2e/+/+/telemetry"
python mqtt_influx_bridge.py

Depois verifique os dados no Data Explorer do InfluxDB ou pela CLI:

influx query 'from(bucket: "sensors")
  |> range(start: -10m)
  |> filter(fn: (r) => r._measurement == "telemetry" and r._field == "temperature")
  |> group(columns: ["line"])
  |> mean()' --org iot --token my-super-secret-token

As decisões de design explicadas

O batching mantém a thread de rede rápida

O write_api com WriteOptions coloca os pontos em buffer e os grava em segundo plano, quando 500 pontos se acumulam ou a cada cinco segundos. O callback on_message nunca espera por uma requisição HTTP, então o paho continua processando keep-alives e mensagens recebidas. Uma gravação síncrona por mensagem funciona para meia dúzia de sensores, mas fica para trás rapidamente quando as taxas crescem. O write_api.close() no encerramento descarrega o que ainda estiver no buffer.

Converter números para float evita conflitos de tipo

O InfluxDB fixa o tipo de um field na primeira vez em que ele é gravado em um shard. Se um dispositivo envia "temperature": 21 uma vez e 21.5 depois, a segunda gravação é rejeitada como conflito de tipo. Converter todo número para float elimina essa categoria inteira de erro.

As tags precisam ter baixa cardinalidade

Só identificadores estáveis pertencem às tags: site (por exemplo, uma planta em Campinas), linha, dispositivo. Nunca coloque timestamps, números de sequência ou mensagens de texto livre em tags, porque cada combinação única cria uma nova série, e a quantidade de séries determina o uso de memória e a velocidade das consultas.

Esquema de topic em vez de posições fixas no código

O TOPIC_SCHEMA descreve a hierarquia uma única vez, então um layout de topic diferente, como um caminho ISA-95 do tipo connectmq/factory1/production/line1/cnc1/telemetry, só precisa de uma nova variável de ambiente: TOPIC_SCHEMA="_/site/area/line/device/@measurement".

Tratando outros formatos de payload

Frotas reais raramente concordam em um único formato. Duas variações aparecem o tempo todo, e ambas cabem em to_point com poucas linhas.

Números puros. Muitos dispositivos simples publicam um único valor por topic, como 21.84 em .../room1/temperature. Nesse caso, o último nível do topic é o nome do field, e não o measurement. Tente float(payload) antes de interpretar o JSON e, se der certo, monte um ponto com um measurement fixo, como sensor, usando o último nível do topic como chave do field.

Objetos aninhados. Gateways industriais costumam enviar estruturas como {"metrics": {"spindle_rpm": 11800, "temp_c": 41.7}}. Achate-as antes do loop de fields, unindo as chaves com sublinhado para que o field vire metrics_spindle_rpm. Mantenha a profundidade do achatamento limitada; payloads muito aninhados geralmente contêm dados de configuração ou diagnóstico que nem deveriam estar em um banco de séries temporais.

Seja o que for que você adicionar, mantenha a regra: mensagens desconhecidas ou malformadas são contadas e ignoradas, nunca podem lançar uma exceção no callback. Um único dispositivo com problema não deve conseguir parar a ponte para todos os outros.

Solução de problemas

Para um tratamento de reconexão mais completo, incluindo buffer offline, veja a lógica robusta de reconexão do paho MQTT. Se preferir o caminho só com configuração, o guia do Telegraf MQTT consumer grava o mesmo modelo de dados sem código.

Checklist de produção

1. Conecte via TLS na porta 8883 com MQTT_TLS=1 e as credenciais do portal do seu broker.
2. Use um token dedicado do InfluxDB restrito a um único bucket.
3. Defina no bucket um período de retenção que corresponda ao tempo que você realmente precisa dos dados brutos.
4. Rode a ponte sob o systemd ou como contêiner com reinício automático.
5. Assine com QoS 1 e um client ID fixo para que um reinício não descarte leituras silenciosamente.
6. Monitore os contadores written e skipped; um aumento repentino de mensagens ignoradas geralmente indica uma mudança de firmware.

Com um broker MQTT gerenciado como a ConnectMQ, o primeiro e o quinto itens são só configuração: copie YOUR_USERNAME, YOUR_PASSWORD e YOUR_CLIENT_ID do portal para o ambiente. A ponte em si não muda. Observe que o InfluxDB 3 usa outro cliente Python; o modelo de dados deste guia continua valendo, mas as chamadas de escrita são diferentes. A documentação do influxdb-client cobre o cliente 2.x usado aqui. Para alternativas do lado dos dispositivos, veja a comparação de bibliotecas de cliente MQTT.

Comece com a ConnectMQ

Teste a ponte primeiro contra o servidor MQTT público (gratuito) com dados simulados; ele serve para testes e protótipos, não para dados de produção. Para dispositivos reais, a ConnectMQ oferece um broker privado com TLS, MQTT 5 e um explorador de topics ao vivo que mostra exatamente quais topics a sua ponte vai receber. Cadastre-se em app.connectmq.com para ter um broker Serverless gratuito, sem cartão de crédito, e leia sobre o modelo de broker MQTT na nuvem se estiver planejando crescer.

Perguntas frequentes

Devo usar Telegraf ou Python para armazenar dados MQTT no InfluxDB?

Use o Telegraf quando a configuração der conta dos seus payloads, porque ele não exige código. Use Python quando precisar de validação, enriquecimento, conversão de unidades ou vários formatos de payload.

Quais níveis do topic MQTT devem virar tags?

Identificadores estáveis, como site, linha e dispositivo. Evite qualquer coisa única por mensagem, porque alta cardinalidade de tags deixa o InfluxDB mais lento.

Por que alguns dos meus pontos são rejeitados?

O motivo mais comum é um conflito de tipo de field, quando um field foi gravado primeiro como inteiro e depois como float ou string. Converta os números para float de forma consistente.

A ponte consegue tratar payloads que não são JSON?

Não da forma como está escrita; ela os ignora e os contabiliza. Estenda to_point para interpretar números puros ou outros formatos se os seus dispositivos os enviarem.