DataOps Industrial: Contextualizando e Governando Dados da Planta

DataOps industrial é a disciplina de transformar dados brutos de tecnologia operacional (TO) em informação contextualizada e confiável e entregá-la a cada consumidor que precisa dela, de forma contínua, e não como um projeto de integração pontual. Neste artigo você vai conhecer as quatro funções de uma camada de DataOps, onde ela fica em uma arquitetura baseada em MQTT e como construir um pequeno serviço de contextualização que republica dados limpos em um namespace compartilhado.

O que DataOps industrial realmente significa

O termo vem do DataOps de TI, que aplica ideias de DevOps (automação, versionamento, monitoramento) a pipelines de dados. Na planta, seja um frigorífico no Paraná ou uma mineradora em Minas Gerais, o problema é diferente. Os dados são gerados por CLPs, sensores e máquinas projetados para controle, não para analytics. As tags não têm unidades, os timestamps vêm de relógios diferentes, máquinas idênticas usam nomes diferentes e ninguém é dono do mapeamento entre elas.

Sem uma camada de DataOps, cada consumidor resolve esses problemas por conta própria. A equipe do historiador escreve um mapeamento, o integrador do MES escreve outro e a equipe de ciência de dados limpa uma exportação em CSV à mão. Cada correção é local e frágil. O DataOps industrial leva esse trabalho para uma única camada gerenciada entre as fontes e os consumidores, de modo que a limpeza é feita uma vez e reaproveitada em todo lugar.

O resultado prático é uma camada de dados da planta em que cada valor chega com seu ativo, unidade, timestamp, qualidade e significado, e em que as mudanças nessa camada são revisadas e versionadas como código. É essa mesma camada que pode alimentar um gêmeo digital da linha sem que ele precise de mapeamentos próprios.

As quatro funções de uma camada de DataOps industrial

Contextualização

Contexto é o que transforma um número em informação. Uma leitura de temperatura precisa saber a qual máquina pertence, de qual parte da máquina vem, qual produto estava rodando e qual turno estava em serviço. Parte do contexto é estática (ativo, localização) e pertence ao caminho do topic ou ao modelo do ativo. Outra parte é dinâmica (ordem atual, receita, modo de operação) e funciona melhor publicada como topics de estado separados, que os consumidores podem combinar com a telemetria. O guia de modelagem de dados IIoT mostra como estruturar esse contexto em tipos e hierarquias.

Normalização

A normalização faz coisas semelhantes parecerem iguais. Dois motores de fabricantes diferentes devem reportar current em ampères, reportar state com a mesma enumeração e usar timestamps epoch em milissegundos. Etapas típicas de normalização incluem escalonamento de valor bruto para unidade de engenharia, conversão de unidades, renomeação, mapeamento de enumerações (códigos de estado do fabricante para um conjunto comum) e alinhamento de timestamps em UTC.

Validação

Os dados de TO falham de maneiras específicas: sensores congelam, valores saltam para os limites da faixa analógica quando um fio se rompe, relógios derivam, gateways reenviam dados em buffer fora de ordem. Uma camada de DataOps deve pegar esses problemas antes que os consumidores os vejam. Você não precisa apagar dados suspeitos; muitas vezes é melhor mantê-los e associar um indicador de qualidade, para que um historiador possa armazená-los enquanto quem treina um modelo pode filtrá-los.

Governança

A governança é a função menos técnica e a mais frequentemente ignorada. Alguém precisa ser dono dos tipos de ativo, aprovar mudanças nos caminhos dos topics e saber quais aplicações dependem de quais dados. Na prática, isso significa manter modelos e mapeamentos sob controle de versão, revisar mudanças e publicar um changelog que os consumidores possam ler.

Onde o DataOps fica em uma arquitetura MQTT

Em uma planta baseada em MQTT, o broker é a espinha dorsal de distribuição e a lógica de DataOps vive em clientes ao redor dele. Um layout comum é este:

PLCs / sensors
     |
Edge gateway (protocol conversion, raw tags)
     |  publishes raw/...
MQTT broker  <---->  DataOps services (context, normalize, validate)
     |  republish to modeled namespace: site/area/line/cell/...
Consumers: historian, MES, dashboards, ML pipelines

O namespace bruto é útil para diagnóstico, mas os consumidores devem assinar apenas o namespace modelado. Essa separação permite corrigir um mapeamento sem mexer no gateway e adicionar consumidores sem mexer no mapeamento. É o mesmo princípio do Unified Namespace: um único lugar para publicar e descobrir o estado atual, estruturado pela hierarquia da planta.

Uma observação honesta sobre ferramentas: o broker roteia mensagens; ele não executa o seu código de transformação. Em um ambiente com ConnectMQ, o serviço de contextualização é um cliente MQTT comum (um processo Python, um fluxo Node-RED ou um produto comercial de DataOps) que assina topics brutos e publica topics modelados. O ConnectMQ fornece o broker MQTT gerenciado, a detecção de topics ISA-95 e a decodificação e republicação de Sparkplug B em topics ISA-95, o que cobre uma etapa comum de normalização sem código próprio.

Construindo um pequeno serviço de contextualização

O serviço em Python a seguir assina mensagens de tags brutas, aplica um arquivo de mapeamento, escalona e valida os valores e os republica em um namespace ISA-95. Ele roda no servidor MQTT público (gratuito) para experimentação; use apenas dados simulados ali, nunca dados de produção.

Primeiro, o mapeamento, que ficaria sob controle de versão:

# mapping.yaml
modelVersion: 1.0.3
tags:
  PLC3.DB10.DBW4:
    path: berlin/line2/filler1/speed
    unit: bpm
    rawMin: 0
    rawMax: 27648
    engMin: 0
    engMax: 200
    validMin: 0
    validMax: 180
  PLC3.DB10.DBW6:
    path: berlin/line2/filler1/temperature
    unit: degC
    rawMin: 0
    rawMax: 27648
    engMin: -20
    engMax: 120
    validMin: 5
    validMax: 90

Depois, o serviço:

import json
import sys
import time
import uuid

import paho.mqtt.client as mqtt
import yaml

BROKER = "broker.connectmq.com"
PREFIX = sys.argv[1] if len(sys.argv) > 1 else "connectmq/demo/" + uuid.uuid4().hex[:8]
STALE_SECONDS = 600

with open("mapping.yaml", encoding="utf-8") as f:
    MAPPING = yaml.safe_load(f)

last_seen = {}  # tag -> (valor bruto, primeira vez em que esse valor foi visto)


def scale(raw, m):
    span = m["rawMax"] - m["rawMin"]
    return m["engMin"] + (raw - m["rawMin"]) * (m["engMax"] - m["engMin"]) / span


def quality(tag, raw, value, m, now):
    prev = last_seen.get(tag)
    if prev is None or prev[0] != raw:
        last_seen[tag] = (raw, now)
    elif now - prev[1] > STALE_SECONDS:
        return "STALE"
    if not (m["validMin"] <= value <= m["validMax"]):
        return "OUT_OF_RANGE"
    return "GOOD"


def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(f"{PREFIX}/raw/#", qos=1)
    print("listening on", f"{PREFIX}/raw/#")


def on_message(client, userdata, msg):
    tag = msg.topic.split("/raw/", 1)[1]
    m = MAPPING["tags"].get(tag)
    if m is None:
        client.publish(f"{PREFIX}/dataops/unmapped", json.dumps({"tag": tag}), qos=0)
        return
    raw = json.loads(msg.payload)
    now = time.time()
    value = round(scale(raw["v"], m), 2)
    out = {
        "value": value,
        "unit": m["unit"],
        "ts": raw.get("ts", int(now * 1000)),
        "quality": quality(tag, raw["v"], value, m, now),
        "modelVersion": MAPPING["modelVersion"],
    }
    client.publish(f"{PREFIX}/{m['path']}", json.dumps(out), qos=1)


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="dataops-" + uuid.uuid4().hex[:8])
client.on_connect = on_connect
client.on_message = on_message
client.connect(BROKER, 1883, keepalive=60)
client.loop_forever()

Instale as dependências com pip install paho-mqtt pyyaml, inicie o serviço com um prefixo de sua escolha e envie a ele um valor bruto:

python dataops.py connectmq/demo/plant42
mosquitto_pub -h broker.connectmq.com -p 1883 -i raw-test-42 \
  -t "connectmq/demo/plant42/raw/PLC3.DB10.DBW4" -m '{"v": 16000, "ts": 1760000000000}'
mosquitto_sub -h broker.connectmq.com -p 1883 -i sub-test-42 -t "connectmq/demo/plant42/berlin/#" -v

O assinante vê uma velocidade escalonada de cerca de 115,7 bpm, com unidade, timestamp, qualidade e versão do modelo. Tags não mapeadas vão para um topic dedicado em vez de desaparecer silenciosamente, e é assim que você descobre novas tags que alguém adicionou a um CLP.

Implantando DataOps industrial: um checklist

1. Comece por um consumidor que esteja sofrendo. Escolha a integração que hoje exige mais limpeza manual.
2. Separe os namespaces bruto e modelado desde o primeiro dia e oriente os consumidores a usar apenas o modelado.
3. Coloque os mapeamentos sob controle de versão e implante-os como código, com revisões.
4. Publique a qualidade, não só os valores. Deixe os consumidores decidirem o que fazer com dados defasados ou fora da faixa.
5. Monitore a própria camada de DataOps. Acompanhe tags não mapeadas, taxas de mensagens e contagens de dados defasados; publique-os também como topics.
6. Deixe explícito quem é responsável por cada tipo de ativo e por cada site.
7. Documente os consumidores para que uma mudança de modelo possa ser confrontada com quem depende dela.

Erros comuns

• Tratar DataOps como uma migração única. As plantas mudam o tempo todo. Se ninguém mantiver o mapeamento, ele se deteriora em poucos meses.
• Limpar os dados dentro de cada dashboard. Lógica escondida em ferramentas de visualização não pode ser reaproveitada nem testada.
• Descartar dados ruins. Apagar valores destrói evidências. Sinalize-os.
• Ignorar o tempo. Sem timestamps de origem e sincronização de relógios, cruzar sinais entre máquinas vira adivinhação. O artigo sobre dados de séries temporais industriais explica por que isso importa para armazenamento e analytics.

Escalando para além de uma planta

Quando vários sites publicam na mesma estrutura, a camada de DataOps se torna o contrato entre eles. Um broker MQTT na nuvem permite que os serviços de cada site publiquem dados modelados em um único lugar, enquanto cotas por plano e isolamento multi-tenant mantêm os ambientes separados. Para o panorama arquitetural mais amplo, o guia de IoT industrial com MQTT mostra como gateways, brokers e consumidores se encaixam, e o artigo sobre historiador MQTT cobre o lado do armazenamento.

Rode sua camada de DataOps no ConnectMQ

Crie um broker Serverless gratuito em app.connectmq.com, sem cartão de crédito. Aponte seu gateway de borda para o namespace bruto, rode seu serviço de contextualização como cliente com YOUR_USERNAME, YOUR_PASSWORD e YOUR_CLIENT_ID e use o explorador de topics em tempo real e a visão de dispositivos ISA-95 do ConnectMQ para confirmar que o namespace modelado está do jeito que o seu modelo diz que deveria estar.

Perguntas frequentes

O que é DataOps industrial?

DataOps industrial é a prática de contextualizar, normalizar, validar e governar dados de TO em uma única camada gerenciada, para que historiadores, MES, dashboards e analytics recebam todos a mesma informação limpa e bem descrita.

DataOps industrial é um produto ou uma prática?

É uma prática que pode ser apoiada por produtos. Você pode implementá-la com plataformas comerciais de DataOps, fluxos Node-RED ou serviços próprios, desde que os mapeamentos sejam versionados, tenham dono e sejam aplicados antes que os dados cheguem aos consumidores.

O broker MQTT faz as transformações?

Não. O broker distribui mensagens; a transformação roda em clientes MQTT que assinam topics brutos e publicam topics modelados. O ConnectMQ, além disso, decodifica Sparkplug B e pode republicá-lo em topics ISA-95.

Como o DataOps industrial se relaciona com um Unified Namespace?

O Unified Namespace é onde os dados modelados são publicados e descobertos; o DataOps é o trabalho que torna os dados nele consistentes e confiáveis. Um namespace sem DataOps tende a virar uma coleção bem organizada de dados bagunçados.