Explorer des données MQTT dans un notebook Jupyter : s'abonner, tracer et sauvegarder

Un notebook Jupyter MQTT est le moyen le plus rapide de comprendre un flux de données que vous n'avez jamais vu : quels topics existent, à quoi ressemblent les payloads et comment les valeurs évoluent dans le temps. Dans ce guide, vous allez vous connecter à un broker depuis un notebook, inspecter les messages bruts, tracer un graphique en direct, sauvegarder des snapshots pour une analyse ultérieure et tout arrêter proprement.

Pourquoi utiliser Jupyter pour explorer MQTT

La plupart des outils MQTT répondent à une question à la fois. Un subscriber en ligne de commande montre les messages bruts, un explorateur de bureau montre l'arborescence des topics et un tableau de bord montre des graphiques. Un notebook combine les trois avec toute la pile data de Python, et garde une trace de ce que vous avez essayé. Lorsque vous héritez d'une flotte d'appareils existante, recevez un nouveau modèle de capteur ou devez déboguer une valeur étrange, un notebook vous permet de passer de « qu'est-ce qui est publié » à « voici un graphique et un jeu de données nettoyé » en une seule session.

Le détail technique clé est que paho-mqtt peut exécuter sa boucle réseau dans un thread d'arrière-plan avec loop_start(). Le noyau du notebook reste libre d'exécuter des cellules pendant que les messages continuent d'arriver. Tout ce guide repose sur ce principe.

Installation

Il vous faut Python 3.9 ou plus récent et un environnement Jupyter : JupyterLab, l'interface notebook classique ou l'éditeur de notebooks de VS Code se comportent tous de la même façon pour ce guide. Installez les paquets depuis le notebook pour qu'ils atterrissent dans l'environnement du noyau :

%pip install "paho-mqtt>=2.0" pandas matplotlib pyarrow

pyarrow n'est nécessaire que pour les snapshots Parquet. Si vous préférez le CSV, vous pouvez vous en passer.

Cellule 1 : configuration

Regroupez tout ce que vous pourriez vouloir modifier dans la première cellule. Un run ID aléatoire sépare vos topics de démonstration sur le serveur MQTT public gratuit, où chacun peut tout lire.

import json, time, uuid, threading, random
from collections import deque
from datetime import datetime, timezone

import pandas as pd
import matplotlib.pyplot as plt
import paho.mqtt.client as mqtt

BROKER = "broker.connectmq.com"
PORT = 1883
RUN_ID = uuid.uuid4().hex[:8]
PREFIX = f"connectmq/demo/{RUN_ID}"
SUBSCRIBE_TO = f"{PREFIX}/#"
print("Topic prefix:", PREFIX)

Si un flux tourne déjà, par exemple celui de notre simulateur d'appareils MQTT, définissez plutôt PREFIX avec le préfixe de ce simulateur et sautez la cellule 3.

Cellule 2 : connecter le subscriber en arrière-plan

# Réexécuter cette cellule ne doit pas laisser traîner un ancien client
if "sub" in globals():
    sub.loop_stop()
    sub.disconnect()

messages = deque(maxlen=50_000)

def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        print("connect failed:", reason_code)
    else:
        client.subscribe(SUBSCRIBE_TO, qos=0)

def on_message(client, userdata, msg):
    messages.append({
        "received": datetime.now(timezone.utc),
        "topic": msg.topic,
        "payload": msg.payload,
        "retain": msg.retain,
    })

sub = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"nb-sub-{RUN_ID}")
sub.on_connect = on_connect
sub.on_message = on_message
sub.connect(BROKER, PORT, keepalive=30)
sub.loop_start()

La garde en haut de la cellule est la ligne la plus importante d'un notebook Jupyter MQTT. Les cellules d'un notebook sont réexécutées en permanence. Sans cette garde, chaque exécution crée un nouveau client avec le même client ID ; le broker déconnecte le précédent pour honorer la session la plus récente, le thread d'arrière-plan de l'ancien client se reconnecte et éjecte le nouveau, et vous vous retrouvez avec deux clients qui se battent dans une boucle sans fin. Les messages apparaissent en double, ou pas du tout, et la cause n'est pas évidente.

Notez que le callback stocke les octets bruts et ne fait aucun parsing. Garder des callbacks triviaux garantit qu'un payload malformé ne pourra jamais faire planter le thread réseau, et vous pouvez décider plus tard comment décoder.

Cellule 3 : générer des données de démonstration (facultatif)

Si vous n'avez aucun flux à explorer, cette cellule démarre un petit publisher dans un autre thread d'arrière-plan. Il simule trois capteurs de température pendant deux minutes, en restant bien en dessous de la limite de 10 messages par seconde par client du serveur MQTT public gratuit.

def demo_publisher(duration_s=120):
    pub = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"nb-pub-{RUN_ID}")
    pub.connect(BROKER, PORT, keepalive=30)
    pub.loop_start()
    state = {"room1": 21.0, "room2": 23.5, "server-rack": 30.0}
    end = time.time() + duration_s
    while time.time() < end:
        for device, value in state.items():
            value += random.gauss(0, 0.2)
            state[device] = value
            payload = {"ts": int(time.time() * 1000), "temperature": round(value, 2)}
            pub.publish(f"{PREFIX}/{device}/telemetry", json.dumps(payload))
        time.sleep(1)
    pub.loop_stop()
    pub.disconnect()

threading.Thread(target=demo_publisher, daemon=True).start()

Cellule 4 : inspecter les messages bruts

Avant d'écrire du code de parsing, regardez ce qui arrive réellement. Les payloads du monde réel ne correspondent souvent pas à ce que promet la documentation.

print(len(messages), "messages so far")
for m in list(messages)[-5:]:
    print(m["topic"], m["payload"][:200])

topics = pd.Series([m["topic"] for m in messages]).value_counts()
topics.head(20)

La sortie de value_counts() vous indique combien de topics distincts existent et l'activité de chacun. C'est souvent le premier signe d'un problème, par exemple un appareil qui publie dix fois plus souvent que ses semblables.

Cellule 5 : transformer les messages en DataFrame

def to_frame(msgs):
    rows = []
    for m in msgs:
        try:
            data = json.loads(m["payload"])
        except (UnicodeDecodeError, json.JSONDecodeError):
            continue
        if not isinstance(data, dict):
            continue
        levels = m["topic"].split("/")
        rows.append({
            "ts": pd.to_datetime(data.get("ts"), unit="ms", utc=True) if "ts" in data else m["received"],
            "device": levels[3] if len(levels) > 3 else m["topic"],
            "temperature": data.get("temperature"),
        })
    df = pd.DataFrame(rows)
    if not df.empty:
        df["temperature"] = pd.to_numeric(df["temperature"], errors="coerce")
        df = df.sort_values("ts")
    return df

df = to_frame(list(messages))
df.groupby("device")["temperature"].describe()

La fonction suppose que le nom de l'appareil est le quatrième niveau du topic, ce qui correspond au publisher de démonstration. Les topics du simulateur comportent un niveau de ligne supplémentaire (<prefix>/<line>/<device>/telemetry) : utilisez alors levels[4], ou mieux, lisez le nom de l'appareil dans le payload lorsqu'il y figure. Adapter cette seule ligne à votre hiérarchie de topics est généralement la seule modification de parsing nécessaire.

describe() par appareil donne le nombre, la moyenne, la dispersion et l'étendue dans une seule table. Des valeurs aberrantes dans min ou max signalent généralement des capteurs défaillants ou des confusions d'unités, comme un appareil qui rapporte des degrés Fahrenheit.

Cellule 6 : tracer des données en direct dans le notebook

Le graphique en direct le plus simple et le plus fiable dans Jupyter redessine une figure Matplotlib statique en boucle et efface la sortie précédente. Il fonctionne dans toutes les interfaces de notebook sans widget supplémentaire.

from IPython.display import clear_output

def live_plot(seconds=60, every=2, window="5min"):
    end = time.time() + seconds
    while time.time() < end:
        df = to_frame(list(messages))
        if not df.empty:
            cutoff = df["ts"].max() - pd.Timedelta(window)
            recent = df[df["ts"] >= cutoff]
            fig, ax = plt.subplots(figsize=(10, 4))
            for device, g in recent.groupby("device"):
                ax.plot(g["ts"], g["temperature"], label=device)
            ax.set_title(f"Live temperature, last {window}")
            ax.set_ylabel("deg C")
            ax.legend(loc="upper left")
            clear_output(wait=True)
            plt.show()
        time.sleep(every)

live_plot(seconds=60)

La cellule bloque pendant la durée indiquée, tandis que le thread d'arrière-plan continue de collecter les messages. Une fois terminée, le dernier graphique reste dans le notebook comme trace. Interrompez le noyau pour arrêter plus tôt ; le subscriber continue de tourner car il vit dans son propre thread.

Cellule 7 : sauvegarder des snapshots pour une analyse ultérieure

Les données en direct disparaissent au redémarrage du noyau. Sauvegardez un snapshot dès que vous voyez quelque chose d'intéressant.

stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
df = to_frame(list(messages))
df.to_parquet(f"mqtt_snapshot_{stamp}.parquet", index=False)

raw = pd.DataFrame(list(messages))
raw["payload"] = raw["payload"].apply(lambda b: b.decode("utf-8", errors="replace"))
raw.to_parquet(f"mqtt_raw_{stamp}.parquet", index=False)
print("saved", len(df), "parsed rows and", len(raw), "raw messages")

Sauvegarder à la fois le DataFrame parsé et les messages bruts est délibéré. Lorsque vous modifierez plus tard votre logique de parsing, vous pourrez la réappliquer au snapshot brut au lieu d'attendre que la même situation se reproduise.

Cellule 8 : nettoyer

sub.loop_stop()
sub.disconnect()

Exécutez cette cellule avant de fermer le notebook. L'arrêt du noyau met aussi fin au thread, mais une déconnexion explicite évite de laisser une session à moitié ouverte sur le broker jusqu'à l'expiration du keep-alive.

Checklist de dépannage

Du notebook à la production

Les notebooks servent à l'exploration. Une fois que vous savez quels champs comptent, déplacez la logique dans un script ou un service : le guide paho-mqtt Python couvre un client autonome, et l'analyse MQTT avec pandas présente le modèle des micro-lots sous forme de programme de longue durée. Si les données doivent être conservées, une base de séries temporelles associée à Grafana est généralement la bonne destination, comme décrit dans l'installation de la stack TIG.

Pour des données réelles, passez du serveur MQTT public gratuit à un broker privé. Sur un broker MQTT managé comme ConnectMQ, il suffit de modifier les lignes de connexion de la cellule 2 : appelez sub.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD"), appelez sub.tls_set(), utilisez client_id="YOUR_CLIENT_ID" et connectez-vous au port 8883. Le certificat Let's Encrypt du broker est reconnu par le magasin système par défaut. Pour comparer les bibliothèques Python pour cette étape, consultez le guide des bibliothèques clientes MQTT, et la documentation Jupyter traite des noyaux et des environnements si %pip se comporte de façon inattendue.

Démarrez avec ConnectMQ

Ouvrez un notebook et pointez-le vers le serveur MQTT public gratuit pour explorer des données de démonstration dès maintenant. Lorsque vous passez aux capteurs réels, ConnectMQ vous fournit un broker privé avec TLS, MQTT 5 et un explorateur de topics en direct dans le portail, qui complète bien l'exploration en notebook. Créez un broker Serverless gratuit sur app.connectmq.com, sans carte bancaire.

Questions fréquentes

paho-mqtt fonctionne-t-il dans un notebook Jupyter ?

Oui. Utilisez loop_start() pour que la boucle réseau s'exécute dans un thread d'arrière-plan. Évitez loop_forever() dans un notebook, car il bloque la cellule et le noyau jusqu'à ce que vous l'interrompiez.

Pourquoi est-ce que je reçois chaque message MQTT en double dans mon notebook ?

Vous avez très probablement réexécuté la cellule de connexion et créé un second client avec son propre abonnement, ou deux clients se disputent le même client ID. Arrêtez et déconnectez l'ancien client avant d'en créer un nouveau.

Puis-je utiliser un client MQTT asyncio dans Jupyter ?

Oui. Jupyter exécute déjà une boucle d'événements : vous pouvez donc faire un await sur un client asynchrone directement dans une cellule. L'approche par thread d'arrière-plan avec paho reste plus simple pour l'exploration interactive.

Quel format utiliser pour les snapshots ?

Parquet préserve les types de colonnes et les horodatages tout en restant compact, ce qui le rend idéal pour une analyse ultérieure. Le CSV est plus portable mais perd les informations de type.