Tutorial Node-RED InfluxDB: armazenando e consultando dados MQTT

Node-RED e InfluxDB juntos formam um pipeline de telemetria pequeno, mas completo: as mensagens MQTT chegam, o Node-RED as organiza e o InfluxDB as armazena como séries temporais que você pode consultar e plotar. Neste tutorial Node-RED InfluxDB você vai rodar o InfluxDB 2 no Docker, gravar leituras MQTT com fields e tags adequados e consultá-las de volta com Flux para desenhar um gráfico. Todas as etapas usam um broker gratuito, então você consegue testar a cadeia inteira em menos de meia hora.

Arquitetura: MQTT na entrada, séries temporais na saída

O pipeline tem três papéis, e mantê-los separados facilita escalar depois:

1. Os dispositivos publicam leituras em JSON em topics MQTT como connectmq/demo/a7f3k2/room1/telemetry.
2. O broker roteia essas mensagens para todos os assinantes. Ele não armazena histórico; essa é a função do banco de dados. Se essa distinção é nova para você, veja em detalhes o que um broker MQTT faz.
3. O Node-RED assina, valida e converte cada mensagem em um ponto do InfluxDB e depois o grava. Um segundo ramo consulta o InfluxDB para os gráficos.

O Node-RED é uma boa escolha quando você também quer roteamento, alertas ou um dashboard no mesmo lugar. Se você só precisa de ingestão em alto volume, o consumidor MQTT do Telegraf é uma alternativa, tratada no guia sobre como armazenar dados MQTT no InfluxDB.

Passo 1: rode o InfluxDB 2 com Docker

A imagem oficial pode inicializar uma organização, um bucket e um token de administrador na primeira execução, o que economiza vários cliques:

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=demo-org \
  -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

Abra http://localhost:8086, faça login e você encontrará o bucket sensors com 30 dias de retenção. Em produção, crie um token de API separado, com acesso de escrita apenas a esse bucket, em vez de dar ao Node-RED o token de administrador. Este tutorial é para o InfluxDB 2.x; o InfluxDB 3 muda o lado das consultas, então verifique as notas de compatibilidade antes de copiar os exemplos em Flux.

Passo 2: publique dados de teste no MQTT

Este script simula duas salas e publica em broker.connectmq.com, o servidor MQTT público gratuito. Troque a7f3k2 por uma string aleatória sua; os topics no servidor público são visíveis para todos, então use-o apenas para testes.

# pip install "paho-mqtt>=2.0"
import json
import random
import time
import uuid

import paho.mqtt.client as mqtt

PREFIX = "connectmq/demo/a7f3k2"
ROOMS = {"room1": 22.0, "room2": 25.0}

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                     client_id=f"influx-sim-{uuid.uuid4().hex[:8]}")
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_start()

try:
    while True:
        for room in ROOMS:
            ROOMS[room] += random.uniform(-0.2, 0.2)
            payload = {
                "temperature": round(ROOMS[room], 2),
                "humidity": round(random.uniform(38, 60), 1),
                "ts": int(time.time() * 1000),
            }
            client.publish(f"{PREFIX}/{room}/telemetry", json.dumps(payload), qos=1)
        time.sleep(5)
except KeyboardInterrupt:
    pass
finally:
    client.loop_stop()
    client.disconnect()

Incluir um timestamp do dispositivo (ts) é uma boa prática: se as mensagens atrasarem, o banco ainda registra quando a leitura foi feita, e não quando ela chegou.

Passo 3: instale o node-red-contrib-influxdb

No Node-RED, abra Manage palette → Install e instale node-red-contrib-influxdb. Você ganha três nodes: influxdb out para gravações individuais, influxdb batch para muitos pontos de uma vez e influxdb in para consultas. Todos compartilham um node de configuração influxdb.

Configure o node de configuração:

Passo 4: mapeie payloads MQTT para fields e tags

O InfluxDB distingue fields (os valores medidos, não indexados) de tags (metadados indexados pelos quais você filtra e agrupa). Acertar isso é a decisão de projeto mais importante de toda a configuração Node-RED InfluxDB:

• Coloque leituras numéricas, como temperatura e umidade, em fields.
• Coloque identificadores de baixa cardinalidade, como sala, linha ou ID do dispositivo, em tags.
• Nunca coloque valores ilimitados (timestamps, IDs aleatórios, texto livre) em tags; cada combinação única de tags cria uma nova série, e o uso de memória cresce junto.

Adicione um node mqtt in assinando connectmq/demo/a7f3k2/+/telemetry, com saída "a parsed JSON object", e depois este function node. O node influxdb out aceita um array de dois objetos: fields primeiro, tags depois.

// Function node "to influx point"
const p = msg.payload;
if (typeof p !== "object" || p === null) {
    return null;
}

const fields = {};
if (typeof p.temperature === "number") fields.temperature = p.temperature;
if (typeof p.humidity === "number") fields.humidity = p.humidity;
if (Object.keys(fields).length === 0) {
    node.warn("No numeric fields in " + msg.topic);
    return null;
}

// connectmq/demo/a7f3k2/room1/telemetry -> room1
const room = msg.topic.split("/")[3];

// Usa o timestamp do dispositivo quando presente (milissegundos)
if (typeof p.ts === "number") {
    fields.time = new Date(p.ts);
}

msg.payload = [fields, { room: room, source: "node-red" }];
msg.measurement = "environment";
return msg;

Ligue-o a um node influxdb out com organização demo-org, bucket sensors e precisão de tempo em milissegundos. Deixe o measurement vazio no node para que msg.measurement seja usado. Faça o deploy, espere um minuto e abra o Data Explorer na interface do InfluxDB: o measurement environment aparece com os fields temperature e humidity e uma tag room.

Se a sua versão do node não aceitar um field time dentro do objeto de fields, remova essas duas linhas; o InfluxDB então usará o horário da gravação.

Passo 5: consulte com Flux e plote o resultado

Consultar é igualmente simples. Um inject node dispara a cada minuto para um function node que monta a consulta Flux:

// Function node "build flux query"
msg.query = `
from(bucket: "sensors")
  |> range(start: -6h)
  |> filter(fn: (r) => r._measurement == "environment" and r._field == "temperature")
  |> aggregateWindow(every: 1m, fn: mean, createEmpty: false)
  |> keep(columns: ["_time", "_value", "room"])
`;
return msg;

Envie-a para um node influxdb in com organização demo-org. O node devolve um array de linhas, cada uma com _time, _value e as colunas de tags. Converta-as em pontos de gráfico, uma série por sala:

// Function node "rows to chart"
const rows = Array.isArray(msg.payload) ? msg.payload : [];
const byRoom = {};

for (const r of rows) {
    if (!byRoom[r.room]) byRoom[r.room] = [];
    byRoom[r.room].push({ x: Date.parse(r._time), y: Math.round(r._value * 100) / 100 });
}

const out = [{ payload: [] }];   // limpa o gráfico antes de redesenhar
for (const room of Object.keys(byRoom)) {
    out.push({ topic: room, payload: byRoom[room] });
}
return [out];

Ligue isso a um ui-chart do Dashboard 2.0 com séries vindas de msg.topic, chave X x e chave Y y. Como os dados agora vêm do banco, o gráfico sobrevive a reinicializações automaticamente. O guia de histórico de gráficos no Node-RED compara essa abordagem com opções mais leves, caso o InfluxDB seja mais do que você precisa, e o tutorial do Dashboard 2.0 explica as configurações do gráfico.

Checklist de produção

1. Use um token dedicado do InfluxDB com acesso de escrita a um bucket para o flow de gravação e acesso de leitura para o flow de consulta.
2. Defina uma política de retenção no bucket que corresponda ao tempo em que você precisa dos dados brutos; reduza a amostragem dos dados mais antigos em um segundo bucket com uma task do InfluxDB se precisar de tendências longas.
3. Mantenha as tags com baixa cardinalidade e consistentes entre os dispositivos.
4. Valide os payloads antes de gravar, como faz o function node, para que um dispositivo com defeito não encha o bucket de strings.
5. Use QoS 1 na assinatura e uma sessão persistente para que reinicializações curtas do Node-RED não percam leituras.
6. Use o node influxdb batch se você grava centenas de pontos por segundo, agrupando as mensagens antes com um join node.
7. Monitore erros de gravação: adicione um catch node restrito ao node influxdb out e encaminhe os erros para um debug node ou um alerta.

Solução de problemas

O conflito de tipo de field merece atenção: o InfluxDB fixa o tipo de um field na primeira gravação dentro de um shard. Se um dispositivo envia 22 e depois 22.5, a divergência entre inteiro e float é rejeitada. Como o JSON não distingue os dois, sensores que enviam números inteiros de vez em quando disparam esse erro. Enviar os valores de forma consistente a partir do firmware (por exemplo, sempre com uma casa decimal) e verificar o tipo do field no Data Explorer após as primeiras gravações evita surpresas depois.

Proteja também o lado MQTT

Um banco de dados com tokens bem cuidados não adianta nada se qualquer pessoa puder publicar leituras falsas nos topics que ele ingere. No servidor público é exatamente essa a situação, e por isso ele serve apenas para testes. Em produção, mude o node de configuração mqtt-broker do Node-RED para um broker privado com TLS na porta 8883 e credenciais. No ConnectMQ, cada broker recebe username, password e client ID gerados, e os client IDs ficam vinculados a essas credenciais, então um cliente qualquer não consegue se passar pelo seu flow de ingestão.

Um broker MQTT na nuvem também simplifica a rede: os dispositivos em campo e o Node-RED ao lado do InfluxDB se conectam de saída ao broker, então você não precisa abrir portas para a rede do banco de dados. Isso vale tanto para um broker MQTT gerenciado como o ConnectMQ quanto para os sensores espalhados em uma fazenda ou subestação. O ConnectMQ em si não grava em bancos de dados; o padrão honesto é exatamente o deste artigo, um assinante Node-RED (ou Telegraf) que grava no InfluxDB. Para o básico sobre brokers, wildcards e nodes de configuração, veja o tutorial Node-RED MQTT.

Comece a armazenar seus dados MQTT com o ConnectMQ

Crie uma conta gratuita em app.connectmq.com no plano Serverless, sem cartão de crédito. Crie um broker, copie as credenciais para o node de configuração do Node-RED e seu pipeline InfluxDB passa a ser alimentado por um fluxo privado e criptografado com TLS, que você pode inspecionar ao vivo no explorador de topics do portal.

Perguntas frequentes

Quais versões do InfluxDB o node-red-contrib-influxdb suporta?

Ele suporta InfluxDB 1.x com InfluxQL, 1.8 com Flux e 2.x com tokens, organizações e buckets. Escolha a versão no node de configuração para que os nodes mostrem as opções corretas.

Devo gravar todas as mensagens MQTT no InfluxDB?

Grave todas as leituras de que você possa precisar depois, mas reduza a amostragem dos dados de alta frequência em um bucket separado para tendências de longo prazo. Se os dispositivos publicam mais rápido do que você precisa, calcule médias no Node-RED antes de gravar.

Qual é a diferença entre fields e tags no InfluxDB?

Fields armazenam os valores medidos e não são indexados. Tags armazenam metadados como sala ou ID do dispositivo, são indexadas e devem ter um número limitado de valores distintos.

Posso usar o Grafana em vez de um gráfico do Node-RED?

Sim. Aponte o Grafana para o mesmo bucket do InfluxDB e use a consulta Flux deste guia. O Node-RED continua fazendo a ingestão MQTT enquanto o Grafana cuida da visualização.