Historiador de Dados MQTT: Capturando Séries Temporais Industriais

Um historiador MQTT é um assinante que escuta os topics MQTT da sua planta e armazena cada valor como série temporal, para que os operadores possam ver tendências, comparar e investigar aquilo que o namespace mostra apenas no momento. Neste guia você vai comparar as principais abordagens de historiador, montar dois pipelines funcionais, Telegraf para InfluxDB e Python para TimescaleDB, e conhecer as decisões de projeto que definem se o seu historiador ainda será confiável daqui a um ano.

O que um historiador MQTT faz

Um historiador de processo clássico faz polling de tags no SCADA ou nos CLPs e as comprime em um arquivo proprietário. Um historiador MQTT inverte o fluxo: dispositivos e gateways publicam em um broker, e o historiador é só mais um assinante. Isso tem consequências úteis:

• Nenhuma carga extra na planta. O historiador nunca faz polling de um CLP; ele lê o que já é publicado.
• Um único ponto de integração. Adicionar uma máquina ao namespace a adiciona automaticamente ao historiador, se a assinatura cobrir o ramo dela.
• Contexto vindo dos topics. Com topics ISA-95, site, área, linha e máquina viram tags ou colunas sem mapeamento manual.

O historiador precisa cuidar de tudo o que o broker não faz: armazenamento de longo prazo, consultas eficientes por intervalo de tempo, downsampling e retenção. Um broker guarda o último valor retido e filas de sessão de curta duração, não meses de histórico. Essa divisão de trabalho é explicada em mais detalhes em Unified Namespace vs data lake.

Comparando abordagens de historiador

Não existe uma única resposta certa. Muitas plantas mantêm um historiador quente com semanas ou meses de dados em alta resolução para operadores e engenheiros e enviam tudo para um lake para analytics de longo prazo.

Dados de exemplo para testes

Os dois pipelines esperam topics como <prefix>/<site>/<area>/<line>/<machine>/telemetry com payloads JSON. Se você tem o simulador de gateway do artigo como construir um Unified Namespace, execute-o. Caso contrário, este loop publica dados simulados no servidor MQTT público (gratuito). A área pública pode ser lida por qualquer pessoa, então use seu próprio prefixo aleatório e apenas valores simulados:

PREFIX="connectmq/demo/h6n3"
while true; do
  TS=$(date -u +%Y-%m-%dT%H:%M:%SZ)
  TEMP=$(awk -v r=$RANDOM 'BEGIN{printf "%.1f", 38 + r % 60 / 10}')
  mosquitto_pub -h broker.connectmq.com -p 1883 -i hist-sim-h6n3 \
    -t "$PREFIX/dallas/packaging/line2/filler/telemetry" \
    -m "{\"ts\":\"$TS\",\"temp_c\":$TEMP,\"speed_bpm\":118}"
  sleep 5
done

Pipeline 1: Telegraf e InfluxDB

O plugin MQTT consumer do Telegraf assina o broker e grava no InfluxDB sem nenhum código próprio. Suba o InfluxDB 2 no Docker com um token de desenvolvimento:

docker run -d --name influxdb -p 8086:8086 \
  -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=plant \
  -e DOCKER_INFLUXDB_INIT_BUCKET=historian \
  -e DOCKER_INFLUXDB_INIT_ADMIN_TOKEN=dev-token-change-me \
  influxdb:2.7

Depois crie o telegraf.conf. O parsing de topics transforma cada nível ISA-95 em uma tag do InfluxDB:

[agent]
  interval = "10s"
  flush_interval = "10s"

[[inputs.mqtt_consumer]]
  servers = ["tcp://broker.connectmq.com:1883"]
  topics = ["connectmq/demo/h6n3/+/+/+/+/telemetry"]
  qos = 1
  client_id = "telegraf-historian-h6n3"
  name_override = "telemetry"
  data_format = "json"
  json_time_key = "ts"
  json_time_format = "2006-01-02T15:04:05Z07:00"

  [[inputs.mqtt_consumer.topic_parsing]]
    topic = "connectmq/demo/h6n3/+/+/+/+/telemetry"
    tags = "_/_/_/site/area/line/machine/_"

[[outputs.influxdb_v2]]
  urls = ["http://localhost:8086"]
  token = "dev-token-change-me"
  organization = "plant"
  bucket = "historian"

Execute com telegraf --config telegraf.conf. Abra a interface do InfluxDB na porta 8086, navegue pelo bucket historian e você verá uma measurement telemetry com as tags site, area, line e machine e campos numéricos como temp_c. O parser JSON mantém os campos numéricos por padrão; liste os campos de texto explicitamente com json_string_fields se precisar deles. A documentação do Telegraf MQTT consumer lista todas as opções.

Para um broker privado do ConnectMQ, altere o servidor para ssl://YOUR_BROKER_HOST:8883, adicione username = "YOUR_USERNAME", password = "YOUR_PASSWORD" e defina client_id como YOUR_CLIENT_ID, já que os client IDs ficam vinculados às credenciais emitidas no portal. Remova o prefixo de demonstração e seus níveis de parsing quando seus topics começarem com o nome da sua empresa.

Pipeline 2: Python e TimescaleDB

Quando você quer SQL e joins com ordens de produção ou registros de lote, o TimescaleDB é uma escolha natural. Suba-o no Docker:

docker run -d --name timescaledb -p 5432:5432 -e POSTGRES_PASSWORD=historian timescale/timescaledb:latest-pg16

O assinante abaixo armazena os dados em formato longo, uma linha por métrica, o que lida bem com máquinas que publicam campos diferentes. Instale as dependências com pip install "paho-mqtt>=2.0" "psycopg[binary]".

import json
import os
from datetime import datetime, timezone

import paho.mqtt.client as mqtt
import psycopg

DSN = os.getenv("PG_DSN", "postgresql://postgres:historian@localhost:5432/postgres")
PREFIX = os.getenv("UNS_PREFIX", "connectmq/demo/h6n3")

conn = psycopg.connect(DSN, autocommit=True)
conn.execute("""
    CREATE TABLE IF NOT EXISTS telemetry (
        time    TIMESTAMPTZ NOT NULL,
        site    TEXT, area TEXT, line TEXT, machine TEXT,
        metric  TEXT NOT NULL,
        value   DOUBLE PRECISION
    )""")
conn.execute("SELECT create_hypertable('telemetry', 'time', if_not_exists => TRUE)")

def on_connect(client, userdata, flags, reason_code, properties):
    print("historian connected:", reason_code)
    client.subscribe(f"{PREFIX}/+/+/+/+/telemetry", qos=1)

def on_message(client, userdata, msg):
    if msg.retain:
        return  # os valores retidos já foram armazenados quando publicados pela primeira vez
    site, area, line, machine, _ = msg.topic[len(PREFIX) + 1:].split("/")
    try:
        data = json.loads(msg.payload)
    except ValueError:
        return
    ts = data.pop("ts", None)
    when = datetime.fromisoformat(ts.replace("Z", "+00:00")) if ts else datetime.now(timezone.utc)
    rows = [(when, site, area, line, machine, k, float(v))
            for k, v in data.items() if isinstance(v, (int, float)) and not isinstance(v, bool)]
    with conn.cursor() as cur:
        cur.executemany(
            "INSERT INTO telemetry (time, site, area, line, machine, metric, value) "
            "VALUES (%s, %s, %s, %s, %s, %s, %s)", rows)

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="pg-historian-h6n3")
client.on_connect = on_connect
client.on_message = on_message
client.connect(os.getenv("MQTT_HOST", "broker.connectmq.com"), 1883, keepalive=60)
client.loop_forever()

Consulte médias de cinco minutos por máquina com SQL padrão mais o time_bucket do Timescale:

SELECT time_bucket('5 minutes', time) AS bucket, line, machine, avg(value) AS avg_temp
FROM telemetry
WHERE metric = 'temp_c' AND time > now() - interval '1 day'
GROUP BY bucket, line, machine
ORDER BY bucket;

Decidindo o que armazenar

Armazenar todos os valores publicados para sempre parece seguro, mas deixa as consultas mais lentas e os custos crescem silenciosamente. Combine uma política de armazenamento com operações e engenharia antes de o historiador entrar em produção:

• Valores de processo, como temperaturas, pressões e velocidades: armazene os dados brutos na taxa publicada por algumas semanas e depois mantenha agregados como mínimo, máximo e média por minuto.
• Contadores, como contagens de peças boas e refugos: armazene cada mudança, porque as diferenças entre leituras é que importam para OEE e throughput.
• Estados e alarmes: armazene cada transição com seu timestamp. Têm baixo volume e são essenciais para a análise de paradas.
• Eventos, como conclusões de lote: armazene todos, de preferência em uma tabela com campos próprios, e não como métricas genéricas.
• Diagnósticos, como uptime do gateway ou intensidade de sinal: mantenha uma retenção mais curta, a menos que esteja investigando um problema específico.

Registrar isso por namespace no seu padrão de topics faz com que a configuração do historiador acompanhe o projeto do namespace, em vez de crescer de forma improvisada.

Decisões de projeto que importam

1. Timestamps de origem em vez do horário de chegada. Armazene o timestamp do payload para que dados em buffer ou atrasados caiam no momento certo. Guarde também o horário de chegada se precisar medir latência.
2. Sessões persistentes. Dê ao historiador um client ID fixo e uma sessão persistente (clean_session=False ou uma expiração de sessão do MQTT 5) para que o broker enfileire mensagens QoS 1 durante um reinício curto.
3. Buffer na borda para interrupções longas. As filas do broker cobrem lacunas breves. Para quedas de WAN de horas, faça buffer no gateway; veja store and forward com MQTT.
4. Tratamento de retained messages. Ao assinar, o broker reenvia os valores retidos. Ignore-os ou deduplique por tempo e métrica para que os reinícios não criem linhas duplicadas.
5. Deadbands e report by exception. Publicar apenas mudanças significativas reduz drasticamente o armazenamento sem perder informação.
6. Retenção e downsampling. Mantenha os dados brutos por um período definido e agregados contínuos ou buckets reduzidos por mais tempo.
7. Escala horizontal. Com as assinaturas compartilhadas do MQTT 5, vários workers do historiador podem compartilhar $share/historian/... para que cada mensagem seja gravada uma única vez. No ConnectMQ, assinaturas compartilhadas estão disponíveis a partir do plano Serverless (Paid).

O ConnectMQ não inclui um conector nativo de banco de dados ou historiador, e não precisa de um para esse padrão: qualquer um dos assinantes acima se conecta como um cliente MQTT comum. Para saber mais sobre a modelagem dos próprios dados, leia sobre dados de séries temporais industriais e, para a arquitetura mais ampla, veja o guia de MQTT para IoT industrial.

Alimente seu historiador MQTT a partir do ConnectMQ

Um historiador é tão completo quanto o fluxo que o alimenta. O ConnectMQ é um broker MQTT gerenciado com MQTT 3.1.1 e 5.0, QoS 1 e 2, sessões persistentes, TLS e um portal que entende ISA-95, para que seu historiador em Telegraf ou Python tenha uma fonte confiável. Crie um broker no plano Serverless (Free) em app.connectmq.com, sem cartão de crédito, ou saiba mais sobre operar um broker MQTT na nuvem para a sua planta.

Perguntas frequentes

Um broker MQTT é um historiador?

Não. Um broker roteia mensagens e guarda o último valor retido de cada topic. Um historiador assina o broker e armazena a série temporal completa para consulta.

Devo escolher InfluxDB ou TimescaleDB?

Escolha InfluxDB com Telegraf para a configuração mais rápida, só com arquivos de configuração, e dashboards no Grafana. Escolha TimescaleDB quando quiser SQL e joins com dados relacionais, como ordens e lotes.

Um historiador comercial consegue ler do MQTT?

Muitos historiadores comerciais oferecem conectores MQTT ou Sparkplug. Verifique se o conector suporta seu formato de payload e sua estrutura de topics e se consegue lidar com births e aliases do Sparkplug, caso você os use.

Como evitar perda de dados quando o historiador reinicia?

Use um client ID fixo com sessão persistente e QoS 1 para que o broker enfileire as mensagens por um curto período, e faça buffer na borda para interrupções mais longas.