Explorar datos MQTT en un Jupyter Notebook: suscribirse, graficar y guardar instantáneas

Un Jupyter Notebook con MQTT es la forma más rápida de entender un flujo de datos que nunca has visto: qué topics existen, qué aspecto tienen los payloads y cómo se comportan los valores con el tiempo. En esta guía te conectarás a un broker desde un notebook, inspeccionarás mensajes en bruto, dibujarás una gráfica en vivo, guardarás instantáneas para analizarlas después y lo cerrarás todo limpiamente.

Por qué usar Jupyter para explorar MQTT

La mayoría de las herramientas MQTT responden a una pregunta cada vez. Un suscriptor de línea de comandos muestra mensajes en bruto, un explorador de escritorio muestra el árbol de topics y un dashboard muestra gráficas. Un notebook combina las tres cosas con todo el stack de datos de Python y además deja constancia de lo que has probado. Cuando heredas una flota de dispositivos existente, recibes un nuevo modelo de sensor o necesitas depurar un valor extraño, un notebook te permite pasar de "qué se está publicando" a "aquí tienes una gráfica y un dataset limpio" en una sola sesión.

El detalle técnico clave es que paho-mqtt puede ejecutar su bucle de red en un hilo en segundo plano con loop_start(). El kernel del notebook queda libre para ejecutar celdas mientras siguen llegando mensajes. Todo en esta guía se apoya en eso.

Configuración inicial

Necesitas Python 3.9 o superior y un entorno Jupyter: JupyterLab, la interfaz clásica de notebook o el editor de notebooks de VS Code se comportan igual para esta guía. Instala los paquetes desde dentro del notebook para que queden en el entorno del kernel:

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

pyarrow solo hace falta para las instantáneas Parquet. Si prefieres CSV, puedes omitirlo.

Celda 1: configuración

Deja en la primera celda todo lo que quizá quieras cambiar. Un run ID aleatorio mantiene separados tus topics de demostración en el servidor MQTT público gratuito, donde cualquiera puede leerlo todo.

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 ya tienes un flujo en marcha, por ejemplo de nuestro simulador de dispositivos MQTT, asigna a PREFIX el prefijo de ese simulador y sáltate la celda 3.

Celda 2: conectar el suscriptor en segundo plano

# Reejecutar esta celda no debe dejar un cliente antiguo activo
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 comprobación del principio es la línea más importante de un Jupyter Notebook con MQTT. Las celdas de un notebook se reejecutan constantemente. Sin esa comprobación, cada ejecución crea otro cliente con el mismo client ID, el broker desconecta el anterior para respetar la sesión más reciente, el hilo en segundo plano del cliente anterior se reconecta y expulsa al nuevo, y acabas con dos clientes peleándose en un bucle infinito. Los mensajes aparecen dos veces, o no aparecen, y la causa no es evidente.

Observa que el callback guarda los bytes en bruto y no interpreta nada. Mantener los callbacks triviales significa que un payload mal formado nunca puede tumbar el hilo de red, y puedes decidir más tarde cómo decodificarlo.

Celda 3: generar datos de demostración (opcional)

Si no tienes ningún flujo que explorar, esta celda arranca un pequeño publicador en otro hilo en segundo plano. Simula tres sensores de temperatura durante dos minutos, muy por debajo del límite del broker público de 10 mensajes por segundo por cliente.

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

Celda 4: inspeccionar los mensajes en bruto

Antes de escribir código de interpretación, mira lo que llega de verdad. Los payloads del mundo real a menudo no son lo que promete la documentación.

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 salida de value_counts() te dice cuántos topics distintos existen y cuánta actividad tiene cada uno. Suele ser la primera señal de un problema, como un dispositivo que publica diez veces más a menudo que sus hermanos.

Celda 5: convertir los mensajes en un 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 función supone que el nombre del dispositivo es el cuarto nivel del topic, lo que coincide con el publicador de demostración. Los topics del simulador tienen un nivel de línea adicional (<prefix>/<line>/<device>/telemetry), así que usa levels[4] en ese caso o, mejor aún, lee el nombre del dispositivo del payload cuando venga incluido. Adaptar esta única línea a tu jerarquía de topics suele ser todo el cambio de interpretación que necesitas.

describe() por dispositivo da recuento, media, dispersión y rango en una sola tabla. Los valores atípicos en min o max suelen apuntar a sensores defectuosos o a confusiones de unidades, como un dispositivo que informa en Fahrenheit.

Celda 6: graficar datos en vivo en el notebook

La gráfica en vivo fiable más sencilla en Jupyter redibuja una figura estática de Matplotlib en un bucle y borra la salida anterior. Funciona en cualquier interfaz de notebook sin widgets adicionales.

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 celda se bloquea durante el tiempo que indiques, mientras el hilo en segundo plano sigue recogiendo mensajes. Al terminar, la última gráfica queda en el notebook como registro. Interrumpe el kernel para parar antes; el suscriptor sigue funcionando porque vive en su propio hilo.

Celda 7: guardar instantáneas para analizarlas después

Los datos en vivo desaparecen cuando se reinicia el kernel. Guarda una instantánea cada vez que veas algo interesante.

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

Guardar tanto el frame interpretado como los mensajes en bruto es deliberado. Cuando más adelante cambies tu lógica de interpretación, podrás volver a ejecutarla sobre la instantánea en bruto en lugar de esperar a que se repita la misma situación.

Celda 8: limpieza

sub.loop_stop()
sub.disconnect()

Ejecuta esto antes de cerrar el notebook. Apagar el kernel también termina el hilo, pero una desconexión explícita evita dejar una sesión medio abierta en el broker hasta que expire el keep-alive.

Lista de comprobación para resolver problemas

Del notebook a producción

Los notebooks son para explorar. Cuando sepas qué campos importan, lleva la lógica a un script o servicio: la guía de paho-mqtt en Python cubre un cliente independiente, y el análisis MQTT con pandas muestra el patrón de microlotes como programa de larga duración. Si hay que conservar los datos, una base de datos de series temporales junto con Grafana suele ser el sitio adecuado, como se describe en la configuración del stack TIG.

Para datos reales, pasa del broker público a uno privado. En un broker MQTT gestionado como ConnectMQ, solo cambias las líneas de conexión de la celda 2: llama a sub.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD"), llama a sub.tls_set(), usa client_id="YOUR_CLIENT_ID" y conéctate al puerto 8883. El certificado de Let's Encrypt del broker es de confianza en el almacén por defecto del sistema. Para comparar bibliotecas de Python para ese paso, consulta la guía de bibliotecas cliente MQTT, y la documentación de Jupyter cubre kernels y entornos si %pip se comporta de forma inesperada.

Empieza con ConnectMQ

Abre un notebook y apúntalo al servidor público de pruebas para explorar datos de demostración ahora mismo. Cuando pases a sensores reales, ConnectMQ te ofrece un broker privado con TLS, MQTT 5 y un explorador de topics en vivo en el portal, que combina muy bien con la exploración en notebooks. Crea un broker Serverless gratuito en app.connectmq.com, sin tarjeta de crédito.

Preguntas frecuentes

¿Funciona paho-mqtt dentro de un Jupyter Notebook?

Sí. Usa loop_start() para que el bucle de red se ejecute en un hilo en segundo plano. Evita loop_forever() en un notebook, porque bloquea la celda y el kernel hasta que lo interrumpas.

¿Por qué recibo cada mensaje MQTT dos veces en mi notebook?

Lo más probable es que hayas reejecutado la celda de conexión y creado un segundo cliente con su propia suscripción, o que dos clientes se estén peleando por el mismo client ID. Detén y desconecta el cliente antiguo antes de crear uno nuevo.

¿Puedo usar un cliente MQTT con asyncio en Jupyter?

Sí. Jupyter ya ejecuta un bucle de eventos, así que puedes hacer await de un cliente asíncrono directamente en una celda. El enfoque de hilo en segundo plano con paho es más sencillo para la exploración interactiva.

¿Qué formato debo usar para las instantáneas?

Parquet conserva intactos los tipos de columna y las marcas de tiempo y es compacto, lo que lo hace ideal para el análisis posterior. CSV es más portable, pero pierde la información de tipos.