Historian de données MQTT : capturer les séries temporelles industrielles

Un historian MQTT est un abonné qui écoute les topics MQTT de votre usine et stocke chaque valeur sous forme de série temporelle, pour que les opérateurs puissent suivre des tendances, comparer et enquêter sur ce que le namespace ne montre que dans l'instant. Dans ce guide, vous comparerez les principales approches d'historian, mettrez en place deux pipelines fonctionnels, Telegraf vers InfluxDB et Python vers TimescaleDB, et découvrirez les choix de conception qui déterminent si votre historian sera encore fiable dans un an.

Ce que fait un historian MQTT

Un historian de process classique interroge les tags du SCADA ou des PLC et les compresse dans une archive propriétaire. Un historian MQTT inverse le flux : les appareils et les passerelles publient vers un broker, et l'historian n'est qu'un abonné de plus. Cela a des conséquences utiles :

• Aucune charge supplémentaire sur l'usine. L'historian n'interroge jamais un PLC ; il lit ce qui est déjà publié.
• Un seul point d'intégration. Ajouter une machine au namespace l'ajoute automatiquement à l'historian si l'abonnement couvre sa branche.
• Le contexte vient des topics. Avec des topics ISA-95, site, zone, ligne et machine deviennent des tags ou des colonnes sans mapping manuel.

L'historian doit prendre en charge tout ce que le broker ne fait pas : stockage à long terme, requêtes efficaces par plage temporelle, sous-échantillonnage et rétention. Un broker conserve la dernière valeur retenue et des files de session de courte durée, pas des mois d'historique. Cette répartition des rôles est détaillée dans Unified Namespace vs data lake.

Comparer les approches d'historian

Il n'existe pas de réponse unique. Beaucoup d'usines conservent un historian chaud avec des semaines ou des mois de données à haute résolution pour les opérateurs et les ingénieurs, et envoient tout vers un lake pour l'analytique à long terme. Ce même historique sert aussi à rejouer et à calibrer un jumeau numérique de la ligne.

Données d'exemple pour les tests

Les deux pipelines attendent des topics de la forme <prefix>/<site>/<area>/<line>/<machine>/telemetry avec des payloads JSON. Si vous avez le simulateur de passerelle de comment construire un Unified Namespace, lancez-le. Sinon, cette boucle publie des données simulées sur le serveur MQTT public gratuit, réservé aux tests et non aux données de production. L'espace public est lisible par tous : utilisez votre propre préfixe aléatoire et uniquement des valeurs simulées.

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 et InfluxDB

Le plugin MQTT consumer de Telegraf s'abonne au broker et écrit dans InfluxDB sans code personnalisé. Démarrez InfluxDB 2 dans Docker avec un jeton de développement :

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

Créez ensuite telegraf.conf. Le parsing des topics transforme chaque niveau ISA-95 en tag 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"

Lancez-le avec telegraf --config telegraf.conf. Ouvrez l'interface InfluxDB sur le port 8086, parcourez le bucket historian et vous verrez une mesure telemetry avec les tags site, area, line et machine et des champs numériques comme temp_c. Le parser JSON conserve par défaut les champs numériques ; listez explicitement les champs texte avec json_string_fields si vous en avez besoin. La documentation du Telegraf MQTT consumer détaille toutes les options.

Pour un broker ConnectMQ privé, remplacez le serveur par ssl://YOUR_BROKER_HOST:8883, ajoutez username = "YOUR_USERNAME", password = "YOUR_PASSWORD" et définissez client_id sur YOUR_CLIENT_ID, car les client IDs sont liés aux identifiants délivrés dans le portail. Supprimez le préfixe de démo et ses niveaux de parsing lorsque vos topics commencent par le nom de votre entreprise.

Pipeline 2 : Python et TimescaleDB

Lorsque vous voulez du SQL et des jointures avec les ordres de fabrication ou les enregistrements de lots, TimescaleDB est un choix naturel. Démarrez-le dans Docker :

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

L'abonné ci-dessous stocke les données au format long, une ligne par mesure, ce qui s'adapte bien aux machines qui publient des champs différents. Installez les dépendances avec 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  # les valeurs retenues ont déjà été stockées lors de leur première publication
    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()

Interrogez les moyennes sur cinq minutes par machine avec du SQL standard et la fonction time_bucket de 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;

Décider de ce qu'il faut stocker

Stocker indéfiniment chaque valeur publiée semble prudent, mais cela ralentit les requêtes et les coûts augmentent sans bruit. Convenez d'une politique de stockage avec l'exploitation et l'ingénierie avant la mise en service de l'historian :

• Les valeurs de process comme les températures, les pressions et les vitesses : stockez les données brutes à la cadence de publication pendant quelques semaines, puis conservez des agrégats comme le minimum, le maximum et la moyenne sur une minute.
• Les compteurs comme les pièces bonnes et rebutées : stockez chaque changement, car ce sont les écarts entre lectures qui comptent pour l'OEE (TRS) et le débit.
• Les états et les alarmes : stockez chaque transition avec son horodatage. Leur volume est faible et ils sont indispensables à l'analyse des arrêts.
• Les événements comme les fins de lot : stockez-les tous, idéalement dans une table avec leurs propres champs plutôt que sous forme de mesures génériques.
• Les diagnostics comme la disponibilité des passerelles ou la qualité du signal : gardez une rétention plus courte, sauf si vous enquêtez sur un problème précis.

Consigner cela par espace de noms dans votre norme de topics permet à la configuration de l'historian de suivre la conception du namespace au lieu de croître au fil de l'eau.

Les choix de conception qui comptent

1. L'horodatage source plutôt que l'heure d'arrivée. Stockez l'horodatage du payload pour que les données mises en tampon ou retardées se placent au bon moment. Conservez aussi l'heure d'arrivée si vous devez mesurer la latence.
2. Des sessions persistantes. Donnez à l'historian un client ID fixe et une session persistante (clean_session=False ou une expiration de session MQTT 5) pour que le broker mette en file les messages QoS 1 pendant un court redémarrage.
3. Une mise en tampon à l'edge pour les longues coupures. Les files du broker couvrent de brèves interruptions. Pour des coupures WAN de plusieurs heures, mettez en tampon sur la passerelle ; voir store and forward avec MQTT.
4. La gestion des retained messages. À l'abonnement, le broker rejoue les valeurs retenues. Ignorez-les ou dédoublonnez sur l'horodatage et la mesure, pour que les redémarrages ne créent pas de lignes en double.
5. Bandes mortes et report by exception. Ne publier que les changements significatifs réduit considérablement le stockage sans perte d'information.
6. Rétention et sous-échantillonnage. Conservez les données brutes pendant une période définie, et des agrégats continus ou des buckets sous-échantillonnés plus longtemps.
7. Montée en charge horizontale. Avec les abonnements partagés de MQTT 5, plusieurs workers d'historian peuvent partager $share/historian/... pour que chaque message ne soit écrit qu'une fois. Sur ConnectMQ, les abonnements partagés sont disponibles à partir de l'offre Serverless payante.

ConnectMQ n'inclut pas de base de données ni de connecteur d'historian intégré, et ce modèle n'en a pas besoin : chacun des abonnés ci-dessus se connecte comme un client MQTT ordinaire. Pour approfondir la modélisation des données elles-mêmes, lisez les séries temporelles industrielles, et pour l'architecture d'ensemble, consultez le guide MQTT pour l'IoT industriel.

Alimentez votre historian MQTT avec ConnectMQ

Un historian n'est complet que si le flux qui l'alimente l'est aussi. ConnectMQ est un broker MQTT managé avec MQTT 3.1.1 et 5.0, QoS 1 et 2, sessions persistantes, TLS et un portail compatible ISA-95, pour que votre historian Telegraf ou Python dispose d'une source fiable. Créez un broker avec l'offre gratuite Serverless sur app.connectmq.com, sans carte bancaire, ou découvrez comment exploiter un broker MQTT cloud pour votre usine.

Questions fréquentes

Un broker MQTT est-il un historian ?

Non. Un broker route les messages et conserve la dernière valeur retenue par topic. Un historian s'abonne au broker et stocke la série temporelle complète pour pouvoir l'interroger.

Faut-il choisir InfluxDB ou TimescaleDB ?

Choisissez InfluxDB avec Telegraf pour la mise en place la plus rapide, par simple configuration, avec des tableaux de bord Grafana. Choisissez TimescaleDB si vous voulez du SQL et des jointures avec des données relationnelles comme les ordres et les lots.

Un historian commercial peut-il lire depuis MQTT ?

De nombreux historians commerciaux proposent des connecteurs MQTT ou Sparkplug. Vérifiez que le connecteur prend en charge votre format de payload et votre structure de topics, et qu'il sait gérer les births et les alias Sparkplug si vous les utilisez.

Comment éviter de perdre des données au redémarrage de l'historian ?

Utilisez un client ID fixe avec une session persistante et la QoS 1 pour que le broker mette brièvement les messages en file, et mettez en tampon à l'edge pour les coupures plus longues.