Node-RED MySQL: gravando dados MQTT no MySQL ou MariaDB com consultas parametrizadas

Node-RED e MySQL formam uma dupla prática quando você quer guardar dados de sensores MQTT em um banco de dados que sua equipe já sabe consultar, fazer backup e usar em relatórios. Neste tutorial você vai criar uma tabela para as leituras, assinar topics MQTT no Node-RED, inserir cada mensagem com consultas parametrizadas, passar para inserts em lote quando a taxa de mensagens crescer e configurar a retenção para que a tabela não cresça para sempre. Tudo funciona da mesma forma no MariaDB.

Por que gravar dados MQTT no MySQL?

Um broker MQTT transporta mensagens; ele não é um historiador. As retained messages entregam o último valor de cada topic, mas não os valores da semana passada. Para montar relatórios, comparar turnos ou alimentar uma ferramenta de BI, você precisa do histórico em algum lugar durável.

Bancos de séries temporais como o InfluxDB são uma escolha natural para telemetria de alta frequência, e falamos disso em Node-RED com InfluxDB. Mesmo assim, MySQL e MariaDB vencem em muitas situações: a empresa já os utiliza, as ferramentas de relatório se conectam a eles nativamente, você precisa cruzar leituras com dados relacionais como ordens de produção ou listas de ativos, ou a taxa de dados é moderada. Algumas centenas de leituras por minuto é um território confortável para uma tabela bem indexada.

Passo 1: criar o banco de dados e a tabela

Se você ainda não tem um servidor de banco de dados, suba o MariaDB no Docker para testes:

docker run -d --name iot-mariadb -p 3306:3306 \
  -e MARIADB_ROOT_PASSWORD=change-root \
  -e MARIADB_DATABASE=iot \
  -e MARIADB_USER=nodered \
  -e MARIADB_PASSWORD=change-me \
  mariadb:11

Depois crie a tabela. Uma linha por leitura, com o topic original guardado para rastreabilidade e o payload completo guardado para qualquer coisa que você não tenha previsto:

CREATE TABLE mqtt_readings (
  id          BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
  received_at DATETIME(3)  NOT NULL,
  topic       VARCHAR(255) NOT NULL,
  device      VARCHAR(64)  NOT NULL,
  metric      VARCHAR(64)  NOT NULL,
  value       DOUBLE       NULL,
  payload     TEXT         NULL,
  INDEX idx_device_metric_time (device, metric, received_at),
  INDEX idx_time (received_at)
);

Armazene os timestamps em UTC e converta nos relatórios. Misturar horário local e mudanças de horário de verão em uma tabela de séries temporais é uma fonte certa de lacunas confusas e horas duplicadas.

Conceda ao usuário do Node-RED apenas o que ele precisa:

GRANT SELECT, INSERT, DELETE ON iot.mqtt_readings TO 'nodered'@'%';

Passo 2: instalar o nó MySQL do Node-RED

Instale o node-red-node-mysql em Menu → Manage palette. Ele adiciona um nó mysql com uma configuração de banco de dados: host, porta (3306), usuário, senha, nome do banco e fuso horário. Defina o fuso horário como Z ou +00:00 para que as datas sejam interpretadas como UTC.

O nó segue um contrato simples:

• msg.topic contém a instrução SQL.
• msg.payload contém os parâmetros: um array para placeholders ? ou um objeto para placeholders nomeados :name.
• O resultado volta em msg.payload: linhas para um SELECT, ou um objeto de resultado com affectedRows e insertId para um INSERT.

Passo 3: assinar o MQTT

Adicione um nó mqtt in conectado ao seu broker. Para testes, use o servidor MQTT público em broker.connectmq.com, porta 1883, um client ID único, e assine connectmq/demo/db91kz/+/+ com QoS 1 e saída em auto-detect. Os dois níveis de wildcard capturam o dispositivo e a métrica, por exemplo connectmq/demo/db91kz/boiler1/temperature. Substitua db91kz por uma string aleatória sua; tudo o que passa pelo broker público fica visível para outras pessoas.

Passo 4: inserir com consultas parametrizadas

Esta é a regra mais importante deste tutorial: nunca monte SQL concatenando payloads MQTT em uma string. Topics e payloads vêm de dispositivos e, em um broker compartilhado ou mal configurado, potencialmente de qualquer pessoa. Um payload como 1); DROP TABLE mqtt_readings; -- deve terminar como texto inofensivo em uma coluna, e não ser executado.

Adicione um nó function entre o nó mqtt in e o nó mysql:

// Topic: connectmq/demo/db91kz/<device>/<metric>
const mqttTopic = msg.topic;
const parts = mqttTopic.split("/");
if (parts.length !== 5) return null;

const device = parts[3].substring(0, 64);
const metric = parts[4].substring(0, 64);

// Aceita números, strings numéricas ou objetos {"value": ...}
let raw = msg.payload;
let value = null;
if (typeof raw === "number") {
    value = raw;
} else if (typeof raw === "string" && raw.trim() !== "" && !isNaN(Number(raw))) {
    value = Number(raw);
} else if (typeof raw === "object" && raw !== null && typeof raw.value === "number") {
    value = raw.value;
}

const payloadText = typeof raw === "string" ? raw : JSON.stringify(raw);
// Formato DATETIME(3) do MySQL em UTC: 2026-10-16 08:15:30.123
const receivedAt = new Date().toISOString().slice(0, 23).replace("T", " ");

msg.topic =
    "INSERT INTO mqtt_readings (received_at, topic, device, metric, value, payload) " +
    "VALUES (?, ?, ?, ?, ?, ?)";
msg.payload = [receivedAt, mqttTopic, device, metric, value, payloadText];
return msg;

Repare que msg.topic é sobrescrito com a instrução SQL, e por isso o código salva antes o topic MQTT em mqttTopic para a coluna topic. Ligue a função ao nó mysql e adicione um nó debug depois dele para ver o insertId de cada linha.

Testando o insert

mosquitto_pub -h broker.connectmq.com -p 1883 -i sim-db91kz -t "connectmq/demo/db91kz/boiler1/temperature" -m "71.4"
mosquitto_pub -h broker.connectmq.com -p 1883 -i sim-db91kz -t "connectmq/demo/db91kz/boiler1/pressure" -m '{"value": 2.3, "unit": "bar"}'
mosquitto_pub -h broker.connectmq.com -p 1883 -i sim-db91kz -t "connectmq/demo/db91kz/boiler1/state" -m "running"

Depois confira a tabela:

SELECT received_at, device, metric, value, payload
FROM mqtt_readings
ORDER BY id DESC
LIMIT 10;

A terceira mensagem é gravada com value igual a NULL e running na coluna payload, que é exatamente o que você quer para estados e outros dados não numéricos.

Passo 5: inserts em lote para taxas de mensagens mais altas

Um INSERT por mensagem funciona bem para algumas mensagens por segundo. Acima disso, cada ida e volta e cada commit se acumulam. O processamento em lote reduz a carga tanto no Node-RED quanto no banco de dados.

Altere a função do Passo 4 para gerar um array de linha em vez de uma instrução SQL (defina msg.payload = [receivedAt, ...] e não mexa em msg.topic) e depois adicione um nó join em modo manual: combine cada msg.payload em um array e envie após 100 mensagens ou após um timeout de 5 segundos. Confira com um nó debug que o resultado é um array de linhas (um array de arrays), e não um único array plano. Em seguida, use esta função:

const rows = msg.payload;            // array de arrays com 6 elementos
if (!Array.isArray(rows) || rows.length === 0) return null;

const placeholders = rows.map(() => "(?, ?, ?, ?, ?, ?)").join(", ");
msg.topic =
    "INSERT INTO mqtt_readings (received_at, topic, device, metric, value, payload) VALUES " +
    placeholders;
msg.payload = rows.flat();
return msg;

Todo valor continua sendo passado como parâmetro, então o lote é tão seguro quanto o insert individual. Se o Node-RED parar enquanto um lote estiver pendente, até cinco segundos de dados são perdidos; para a maior parte do monitoramento isso é aceitável, e para qualquer caso mais rigoroso, mantenha os inserts individuais.

Passo 6: retenção e consultas úteis

Decida por quanto tempo as leituras brutas serão mantidas. Uma limpeza diária simples no Node-RED usa um nó inject configurado para repetir em um horário fixo, seguido de um nó function:

msg.topic = "DELETE FROM mqtt_readings WHERE received_at < ? LIMIT 50000";
const cutoff = new Date(Date.now() - 90 * 24 * 3600 * 1000);
msg.payload = [cutoff.toISOString().slice(0, 23).replace("T", " ")];
return msg;

O LIMIT mantém cada exclusão pequena para não travar a tabela por muito tempo; rode com mais frequência se um dia gerar mais linhas do que isso. O event scheduler do MySQL pode fazer o mesmo trabalho dentro do banco, se você preferir.

Duas consultas que você vai usar o tempo todo:

-- Média por hora de cada dispositivo nas últimas 24 horas
SELECT device,
       DATE_FORMAT(received_at, '%Y-%m-%d %H:00') AS hour,
       AVG(value) AS avg_value,
       MAX(value) AS max_value
FROM mqtt_readings
WHERE metric = 'temperature'
  AND received_at >= UTC_TIMESTAMP() - INTERVAL 24 HOUR
GROUP BY device, hour
ORDER BY device, hour;

-- Dispositivos que não reportam há 15 minutos
SELECT device, MAX(received_at) AS last_seen
FROM mqtt_readings
GROUP BY device
HAVING last_seen < UTC_TIMESTAMP() - INTERVAL 15 MINUTE;

A segunda consulta é uma forma barata de encontrar dispositivos silenciosos. Para ser avisado em vez de precisar olhar, combine-a com o fluxo de alerta por e-mail do Node-RED. Para exibir o histórico em um dashboard, o guia de histórico em gráficos do Node-RED mostra como consultá-lo e plotá-lo.

Solução de problemas em fluxos Node-RED MySQL

Configuração para produção

Para dados reais, saia do broker público. Crie um broker privado na ConnectMQ e configure o nó de broker do Node-RED com a porta 8883, TLS ativado e os valores YOUR_USERNAME, YOUR_PASSWORD e YOUR_CLIENT_ID do portal. Use uma sessão persistente (clean session desativado) com QoS 1 para que reinícios rápidos do Node-RED não percam mensagens que o broker ainda consegue entregar; no plano Serverless gratuito o QoS vai até 1, que de qualquer forma é o nível certo para registro de dados.

Essa divisão de responsabilidades é proposital. A ConnectMQ é um broker MQTT gerenciado que entrega mensagens; ela não grava em bancos de dados por conta própria. O Node-RED é o cliente voltado ao banco de dados, o que mantém o seu schema, a retenção e o SQL inteiramente sob o seu controle. O tutorial Node-RED MQTT cobre as demais configurações do nó de broker.

Registre seus dados MQTT com Node-RED e ConnectMQ

Crie um broker no plano Serverless gratuito em app.connectmq.com, sem cartão de crédito, conecte o Node-RED com TLS e comece a montar um histórico que você pode consultar com SQL puro, seja de uma caldeira em um frigorífico ou de bombas em uma estação de saneamento. O explorador de topics ao vivo do portal ajuda a confirmar topics e formatos de payload antes de projetar a tabela em torno deles. Se você ainda está avaliando opções de hospedagem, nosso guia de broker MQTT na nuvem compara as abordagens.

Perguntas frequentes

Como passo parâmetros para o nó MySQL do Node-RED?

Coloque o SQL com placeholders ? em msg.topic e um array de valores em msg.payload. O nó também aceita placeholders nomeados :name quando msg.payload é um objeto.

Isso funciona com MariaDB?

Sim. O node-red-node-mysql se conecta tanto ao MySQL quanto ao MariaDB, e a tabela, os inserts e as consultas deste tutorial rodam sem alterações em qualquer um dos dois.

Devo usar MySQL ou InfluxDB para dados MQTT?

Use MySQL ou MariaDB quando precisar de joins com dados relacionais, de ferramentas de relatório existentes ou trabalhar com taxas de dados moderadas. Use um banco de séries temporais para telemetria de alta frequência e retenção longa com downsampling.

A ConnectMQ consegue armazenar mensagens diretamente em um banco de dados?

Não. A ConnectMQ é um broker MQTT sem integrações com bancos de dados. Rode o Node-RED ou outro assinante que se conecte ao broker e grave no banco de dados, como mostrado neste tutorial.