Dados de séries temporais industriais são o fluxo de medições com timestamp que máquinas, sensores e controladores produzem a cada segundo de cada turno. Tratá-los bem significa escolher com que frequência amostrar, o que descartar, como aplicar o timestamp e onde guardá-los. Este artigo percorre cada uma dessas decisões e traz código funcional para aplicar compressão deadband a dados MQTT antes que cheguem a um historiador.
O que torna os dados de séries temporais industriais diferentes
Todas as séries temporais têm o mesmo formato: um timestamp, um identificador e um valor. Os dados industriais acrescentam várias complicações que os dados de monitoramento de TI raramente têm.
- Taxas de variação muito diferentes. O nível de um reservatório em uma estação de saneamento muda lentamente ao longo de horas; um sinal de vibração muda milhares de vezes por segundo. Uma única política de amostragem não serve para os dois.
- Requisitos de retenção longos. Investigações de qualidade, registros regulatórios, treinamento de modelos e um gêmeo digital do processo frequentemente exigem anos de histórico, não semanas.
- A qualidade importa tanto quanto o valor. Um valor de um sensor desconectado não é uma medição. Sistemas industriais carregam um indicador de qualidade junto com cada ponto.
- Dados de eventos e estados misturados a dados analógicos. Estados de máquina, alarmes, IDs de lote e trocas de receita são esparsos e precisam ser cruzados com sinais analógicos densos.
- Relógios que você não controla. CLPs, gateways e servidores têm cada um o seu relógio, e eles derivam.
Acertar isso desde o início é muito mais barato do que consertar anos de histórico inconsistente depois.
Escolhendo as taxas de amostragem
A taxa de amostragem deve seguir a física do sinal e a pergunta que você quer responder, não o tempo de varredura padrão do CLP.
| Tipo de sinal | Comportamento típico | Abordagem prática |
|---|---|---|
| Temperaturas, níveis, pressões lentas | Muda em segundos a minutos | Amostrar a cada 1–10 s, comprimir com deadband |
| Vazões, corrente de motor, velocidades | Muda em frações de segundo a segundos | Amostrar a cada 100 ms–1 s, comprimir com deadband |
| Estado de máquina, alarmes, contadores | Eventos discretos | Publicar apenas na mudança (report by exception) |
| Vibração, acústica, qualidade de energia | Muda em milissegundos | Amostrar em kHz na borda, publicar features (RMS, pico, bandas) |
Duas regras ajudam. Primeiro, o princípio de Nyquist: para enxergar uma variação, amostre pelo menos duas vezes mais rápido que ela e, na prática, várias vezes mais rápido. Segundo, não transmita formas de onda brutas de alta frequência continuamente pelo MQTT. Calcule features na borda e publique as features; envie snapshots brutos apenas sob demanda ou em eventos.
Para dados discretos, report by exception é o padrão natural. O artigo sobre report by exception explica como ele reduz o tráfego sem perder informação.
Compressão: guardar o sinal, descartar o ruído
Compressão em historiadores industriais geralmente significa filtragem com perdas na ingestão, não arquivos zip. O objetivo é armazenar menos pontos e ainda conseguir reconstruir o sinal dentro de uma tolerância conhecida.
Filtragem por deadband (exceção)
O método mais simples: armazenar um novo ponto apenas quando o valor difere do último valor armazenado por mais que um limiar. Um deadband de 0,2 degC em uma temperatura significa que flutuações abaixo de 0,2 degC não são armazenadas. A maioria das implementações também armazena um ponto após um tempo máximo, mesmo que nada tenha mudado (um heartbeat), para que os consumidores consigam distinguir um sinal estável de um sinal morto.
Compressão swinging door
O swinging door trending, usado em muitos historiadores comerciais, mantém um ponto apenas quando uma reta a partir do último ponto armazenado já não consegue representar os valores intermediários dentro da tolerância. Ele preserva inclinações muito melhor que um deadband simples, então rampas são armazenadas com poucos pontos em vez de muitos.
Escolhendo as tolerâncias
Defina a tolerância em relação à exatidão do sensor, não a um percentual arbitrário. Filtrar abaixo do nível de ruído do sensor armazena ruído; filtrar muito acima dele perde mudanças reais do processo. Documente a tolerância por tipo de tag no seu modelo de dados IIoT, para que ela seja revisada como qualquer outro atributo do modelo.
Compressão deadband em dados MQTT com Python
Este assinante lê telemetria JSON do MQTT, aplica um deadband com heartbeat e grava os pontos que sobrevivem como line protocol do InfluxDB na saída padrão, que você pode redirecionar para um arquivo ou encaminhar a um writer. Ele usa o servidor MQTT público (gratuito) com um simulador de demonstração embutido; esse servidor serve para testes e protótipos, não para dados de produção.
import json
import math
import random
import threading
import time
import uuid
import paho.mqtt.client as mqtt
BROKER = "broker.connectmq.com"
PREFIX = f"connectmq/demo/{uuid.uuid4().hex[:8]}/plant1/line1"
DEADBAND = {"temperature": 0.2, "pressure": 0.05}
HEARTBEAT_S = 60
last_stored = {} # topic -> (valor, ts_segundos)
def keep(topic, metric, value, ts):
prev = last_stored.get(topic)
if prev is None:
return True
band = DEADBAND.get(metric, 0.0)
return abs(value - prev[0]) > band or ts - prev[1] >= HEARTBEAT_S
def on_connect(client, userdata, flags, reason_code, properties):
client.subscribe(f"{PREFIX}/+/+", qos=1)
def on_message(client, userdata, msg):
asset, metric = msg.topic.split("/")[-2:]
p = json.loads(msg.payload)
ts = p["ts"] / 1000
if keep(msg.topic, metric, p["value"], ts):
last_stored[msg.topic] = (p["value"], ts)
print(f'{metric},asset={asset},quality={p["quality"]} value={p["value"]} {int(p["ts"] * 1e6)}')
def simulate(pub):
t0 = time.time()
while True:
t = time.time() - t0
temp = 60 + 5 * math.sin(t / 120) + random.gauss(0, 0.05)
pres = 2.5 + random.gauss(0, 0.01)
ts = int(time.time() * 1000)
for metric, val in [("temperature", temp), ("pressure", pres)]:
body = {"value": round(val, 3), "ts": ts, "quality": "GOOD"}
pub.publish(f"{PREFIX}/pump1/{metric}", json.dumps(body), qos=0)
time.sleep(1)
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="tsdb-" + uuid.uuid4().hex[:8])
client.on_connect = on_connect
client.on_message = on_message
client.connect(BROKER, 1883, keepalive=60)
threading.Thread(target=simulate, args=(client,), daemon=True).start()
client.loop_forever()
Deixe rodar por alguns minutos e conte as linhas: a pressão, que só apresenta ruído abaixo do seu deadband, gera cerca de um ponto por minuto (o heartbeat), enquanto a temperatura gera pontos à medida que a senoide se move. Esse é exatamente o comportamento que você quer de um historiador: denso onde o processo se move, esparso onde não se move.
Timestamps e ordenação
Aplique sempre o timestamp na origem, o mais perto possível da medição, em UTC e com milissegundos. O horário de chegada ao broker não é o horário da medição, especialmente quando um gateway de borda faz buffer dos dados durante uma queda de rede e os encaminha depois. O padrão store and forward depende de timestamps de origem para colocar os dados atrasados no lugar certo.
Sincronize os relógios com NTP ou PTP nos gateways e, quando possível, nos controladores. Conte com chegadas fora de ordem e garanta que seu armazenamento aceite gravações no passado. Alguns bancos de séries temporais lidam bem com gravações atrasadas; outros as penalizam.
Opções de armazenamento para dados de séries temporais industriais
| Opção | Pontos fortes | Atenção a |
|---|---|---|
| Historiador de processo comercial | Compressão, qualidade, modelos de ativos e retenção longa nativos | Licenciamento, acesso proprietário |
| Banco de séries temporais (InfluxDB, TimescaleDB, QuestDB) | Interfaces abertas, consultas SQL ou Flux, boas taxas de escrita | Você mesmo projeta compressão, retenção e downsampling |
| Data lake (Parquet em object storage) | Armazenamento de longo prazo barato, ideal para treinar ML | Não serve para consultas em tempo real |
| Combinação em camadas | Dados quentes no TSDB, dados frios no lake | Dois sistemas para operar |
Muitas plantas combinam um historiador ou TSDB para dados recentes com um lake para analytics de longo prazo. O artigo sobre historiador MQTT mostra como construir o lado do historiador diretamente a partir de topics MQTT.
Levando os dados do broker para o armazenamento
Um broker é um roteador, não um banco de dados: as mensagens passam por ele e as retained messages guardam apenas o último valor de cada topic. Para armazenar o histórico, você roda um writer que assina e insere. O Telegraf é uma escolha comum:
[[inputs.mqtt_consumer]]
servers = ["ssl://broker.connectmq.com:8883"]
topics = ["connectmq/plant1/#"]
client_id = "YOUR_CLIENT_ID"
username = "YOUR_USERNAME"
password = "YOUR_PASSWORD"
qos = 1
data_format = "json"
json_time_key = "ts"
json_time_format = "unix_ms"
tag_keys = ["quality"]
[[inputs.mqtt_consumer.topic_parsing]]
topic = "connectmq/+/+/+/+"
tags = "_/site/line/asset/_"
measurement = "_/_/_/_/measurement"
[[outputs.influxdb_v2]]
urls = ["http://localhost:8086"]
token = "YOUR_INFLUX_TOKEN"
organization = "plant"
bucket = "telemetry"
Com o ConnectMQ como broker MQTT gerenciado, o Telegraf é simplesmente mais um cliente autenticado via TLS na porta 8883. Use QoS 1 na assinatura do writer para que uma desconexão breve não descarte dados silenciosamente e mantenha a sessão do writer persistente se quiser mensagens enfileiradas enquanto ele reinicia.
Checklist para dados de séries temporais industriais
- Classifique cada sinal: analógico lento, analógico rápido, discreto ou de alta frequência.
- Defina amostragem e deadband por classe, com base na exatidão do sensor.
- Adicione um heartbeat para distinguir sinais estáveis de sinais mortos.
- Aplique o timestamp na origem, em UTC e com milissegundos.
- Publique a qualidade junto com cada valor.
- Calcule features na borda para sinais em kHz.
- Escolha o armazenamento quente e frio e defina retenção e downsampling.
- Teste gravações atrasadas e fora de ordem antes de entrar em produção.
Para a arquitetura em torno de tudo isso, veja o guia de IoT industrial com MQTT.
Comece com o ConnectMQ
Crie um broker Serverless gratuito em app.connectmq.com, sem cartão de crédito, e conecte seu gateway e um writer em Telegraf ou Python com as credenciais geradas. O explorador de topics em tempo real do ConnectMQ ajuda a conferir taxas de publicação e payloads antes de você se comprometer com uma política de amostragem e compressão.
Perguntas frequentes
O que são dados de séries temporais industriais?
São dados de medições e eventos com timestamp produzidos por equipamentos industriais, como temperaturas, pressões, velocidades, estados e contadores. Normalmente são armazenados com um indicador de qualidade e mantidos por longos períodos para operações, qualidade e analytics.
Qual é um bom valor de deadband?
Comece perto da exatidão ou do nível de ruído do sensor para aquela medição e depois confira se as mudanças importantes do processo continuam visíveis nos dados armazenados. Documente o valor por tipo de tag e revise-o quando os sensores mudarem.
Um broker MQTT pode armazenar histórico de séries temporais?
Não. Um broker roteia mensagens e guarda no máximo o último valor retido de cada topic. Armazene o histórico com um assinante, como o Telegraf, um conector de historiador ou um writer próprio.
Devo transmitir dados brutos de vibração via MQTT?
Normalmente não de forma contínua. Calcule features como RMS, pico e energias por banda na borda e publique-as; envie snapshots da forma de onda bruta apenas sob demanda ou quando um evento os disparar.
