DataOps industriel : contextualiser et gouverner les données d'usine

Le DataOps industriel est la discipline qui consiste à transformer les données brutes des technologies opérationnelles (OT) en informations contextualisées et fiables, puis à les livrer à chaque consommateur qui en a besoin, en continu plutôt que dans le cadre d'un projet d'intégration ponctuel. Dans cet article, vous découvrirez les quatre missions d'une couche DataOps, sa place dans une architecture basée sur MQTT et comment construire un petit service de contextualisation qui republie des données propres dans un namespace partagé.

Ce que recouvre réellement le DataOps industriel

Le terme est emprunté au DataOps de l'IT, qui applique les idées du DevOps (automatisation, versioning, supervision) aux pipelines de données. À l'usine, le problème est différent. Les données sont générées par des PLC, des capteurs et des machines conçus pour le contrôle-commande, pas pour l'analytique. Les tags n'ont pas d'unités, les horodatages proviennent d'horloges différentes, des machines identiques portent des noms différents, et personne n'est responsable de la correspondance entre eux.

Sans couche DataOps, chaque consommateur résout ces problèmes de son côté. L'équipe historian écrit un mapping, l'intégrateur MES en écrit un autre, et l'équipe data science nettoie un export CSV à la main. Chaque correction est locale et fragile. Le DataOps industriel déplace ce travail dans une couche gérée unique, entre les sources et les consommateurs, pour que le nettoyage soit fait une fois et réutilisé partout.

Le résultat concret est une couche de données d'usine où chaque valeur arrive avec son actif, son unité, son horodatage, sa qualité et sa signification, et où les modifications de cette couche sont relues et versionnées comme du code. C'est aussi le socle de données dont a besoin tout jumeau numérique de l'usine.

Les quatre missions d'une couche de DataOps industriel

Contextualisation

Le contexte est ce qui transforme un nombre en information. Une mesure de température doit savoir à quelle machine elle appartient, à quelle partie de la machine, quel produit était en cours et quelle équipe était de service. Une partie du contexte est statique (actif, emplacement) et a sa place dans le chemin du topic ou dans le modèle d'actif. Une autre est dynamique (ordre en cours, recette, mode de fonctionnement) et se publie de préférence sur des topics d'état séparés que les consommateurs peuvent croiser avec la télémétrie. Le guide de modélisation des données IIoT explique comment structurer ce contexte en types et en hiérarchies.

Normalisation

La normalisation fait se ressembler ce qui est similaire. Deux moteurs de fournisseurs différents doivent tous deux remonter current en ampères, tous deux remonter state avec la même énumération, et tous deux utiliser des horodatages epoch en millisecondes. Les étapes de normalisation typiques comprennent la mise à l'échelle des valeurs brutes en unités physiques, la conversion d'unités, le renommage, le mapping d'énumérations (codes d'état fournisseur vers un ensemble commun) et l'alignement des horodatages sur UTC.

Validation

Les données OT défaillent de manières bien précises : des capteurs se figent, des valeurs sautent aux limites de la plage analogique quand un fil casse, des horloges dérivent, des passerelles rejouent des données mises en tampon dans le désordre. Une couche DataOps doit détecter ces cas avant que les consommateurs ne les voient. Il n'est pas nécessaire de supprimer les données suspectes ; il vaut souvent mieux les conserver en leur associant un indicateur de qualité, pour qu'un historian puisse les stocker tandis qu'un entraînement de modèle peut les filtrer.

Gouvernance

La gouvernance est la mission la moins technique et celle que l'on néglige le plus souvent. Quelqu'un doit être responsable des types d'actifs, approuver les modifications des chemins de topics et savoir quelles applications dépendent de quelles données. En pratique, cela signifie conserver les modèles et les mappings sous contrôle de version, relire les changements et publier un journal des modifications lisible par les consommateurs.

La place du DataOps dans une architecture MQTT

Dans une usine basée sur MQTT, le broker est la colonne vertébrale de distribution et la logique DataOps vit dans des clients qui gravitent autour. Une organisation courante ressemble à ceci :

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

Le namespace brut est utile au dépannage, mais les consommateurs ne doivent s'abonner qu'au namespace modélisé. Cette séparation permet de corriger un mapping sans toucher à la passerelle, et d'ajouter des consommateurs sans toucher au mapping. C'est le même principe que le Unified Namespace : un endroit unique pour publier et découvrir l'état courant, structuré selon la hiérarchie de l'usine.

Un mot d'honnêteté sur l'outillage : le broker route les messages ; il n'exécute pas votre code de transformation. Dans une configuration ConnectMQ, le service de contextualisation est un client MQTT ordinaire (un processus Python, un flow Node-RED ou un produit DataOps commercial) qui s'abonne aux topics bruts et publie les topics modélisés. ConnectMQ fournit le broker MQTT managé, la détection des topics ISA-95 ainsi que le décodage Sparkplug B et sa republication dans des topics ISA-95, ce qui couvre une étape de normalisation courante sans code personnalisé.

Construire un petit service de contextualisation

Le service Python suivant s'abonne aux messages de tags bruts, applique un fichier de mapping, met à l'échelle et valide les valeurs, puis les republie dans un namespace ISA-95. Il fonctionne avec le serveur MQTT public gratuit pour l'expérimentation ; n'y utilisez que des données simulées.

D'abord, le mapping, qui aurait sa place sous contrôle de version :

# 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

Puis le service :

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 -> (valeur brute, première fois que cette valeur a été vue)


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()

Installez les dépendances avec pip install paho-mqtt pyyaml, démarrez le service avec le préfixe de votre choix et envoyez-lui une valeur brute :

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

L'abonné reçoit une vitesse mise à l'échelle d'environ 115,7 bpm, avec unité, horodatage, qualité et version du modèle. Les tags non mappés arrivent sur un topic dédié au lieu de disparaître en silence : c'est ainsi que vous découvrez les nouveaux tags que quelqu'un a ajoutés à un PLC.

Déployer le DataOps industriel : une checklist

1. Commencez par le consommateur qui souffre le plus. Choisissez l'intégration qui exige aujourd'hui le plus de nettoyage manuel.
2. Séparez les namespaces brut et modélisé dès le premier jour, et demandez aux consommateurs de n'utiliser que le namespace modélisé.
3. Placez les mappings sous contrôle de version et déployez-les comme du code, avec des revues.
4. Publiez la qualité, pas seulement les valeurs. Laissez les consommateurs décider quoi faire des données obsolètes ou hors plage.
5. Supervisez la couche DataOps elle-même. Suivez les tags non mappés, les débits de messages et le nombre de valeurs obsolètes ; publiez-les aussi sous forme de topics.
6. Rendez la responsabilité explicite pour chaque type d'actif et chaque site.
7. Documentez les consommateurs pour pouvoir confronter une modification du modèle à ceux qui en dépendent.

Erreurs courantes

• Traiter le DataOps comme une migration ponctuelle. Les usines changent sans cesse. Si personne ne maintient le mapping, il se dégrade en quelques mois.
• Nettoyer les données dans chaque tableau de bord. Une logique cachée dans des outils de visualisation ne peut être ni réutilisée ni testée.
• Supprimer les mauvaises données. Effacer des valeurs détruit des preuves. Signalez-les plutôt.
• Négliger le temps. Sans horodatages source ni synchronisation des horloges, croiser des signaux entre machines relève de la devinette. L'article sur les séries temporelles industrielles explique pourquoi c'est essentiel pour le stockage et l'analytique.

Passer à l'échelle au-delà d'une usine

Quand plusieurs sites publient dans la même structure, la couche DataOps devient le contrat qui les lie. Un broker MQTT cloud permet aux services de chaque site de publier leurs données modélisées en un seul endroit, tandis que les quotas par offre et l'isolation multi-tenant maintiennent les environnements séparés. Pour la vue architecturale d'ensemble, le guide de l'IoT industriel avec MQTT montre comment passerelles, brokers et consommateurs s'articulent, et l'article sur l'historian MQTT couvre le volet stockage.

Faites tourner votre couche DataOps sur ConnectMQ

Créez un broker Serverless gratuit sur app.connectmq.com, sans carte bancaire. Pointez votre passerelle edge vers le namespace brut, exécutez votre service de contextualisation comme client avec YOUR_USERNAME, YOUR_PASSWORD et YOUR_CLIENT_ID, et utilisez l'explorateur de topics en direct et la vue des appareils ISA-95 de ConnectMQ pour confirmer que le namespace modélisé correspond à ce que prévoit votre modèle.

Questions fréquentes

Qu'est-ce que le DataOps industriel ?

Le DataOps industriel consiste à contextualiser, normaliser, valider et gouverner les données OT dans une couche gérée unique, afin que les historians, le MES, les tableaux de bord et l'analytique reçoivent tous les mêmes informations propres et bien décrites.

Le DataOps industriel est-il un produit ou une pratique ?

C'est une pratique que des produits peuvent soutenir. Vous pouvez la mettre en œuvre avec des plateformes DataOps commerciales, des flows Node-RED ou vos propres services, du moment que les mappings sont versionnés, ont un responsable et sont appliqués avant que les données n'atteignent les consommateurs.

Le broker MQTT effectue-t-il les transformations ?

Non. Le broker distribue les messages ; la transformation s'exécute dans des clients MQTT qui s'abonnent aux topics bruts et publient les topics modélisés. ConnectMQ décode en plus Sparkplug B et peut le republier dans des topics ISA-95.

Quel est le lien entre DataOps industriel et Unified Namespace ?

Le Unified Namespace est l'endroit où les données modélisées sont publiées et découvertes ; le DataOps est le travail qui rend ces données cohérentes et fiables. Un namespace sans DataOps tend à devenir une collection bien rangée de données désordonnées.