Análisis de datos MQTT en tiempo real con Python y pandas

Combinar MQTT y pandas te ofrece una forma rápida de analizar datos IoT en vivo con Python sin montar antes una base de datos. En este tutorial te suscribirás a un broker con paho-mqtt, acumularás los mensajes de forma segura, los convertirás en DataFrames cada pocos segundos y calcularás estadísticas móviles, remuestreos por minuto y señales sencillas de anomalías.

Cuándo MQTT y pandas son la combinación adecuada

pandas es una biblioteca orientada a lotes. Es excelente operando sobre una tabla de filas a la vez y mala recibiendo actualizaciones fila a fila. MQTT es justo lo contrario: los mensajes llegan de uno en uno, a intervalos irregulares, en un hilo de red que no controlas. El truco es poner un pequeño búfer entre ambos y dejar que pandas trabaje con microlotes.

Este patrón encaja bien cuando quieres explorar un flujo de datos nuevo, prototipar un cálculo antes de llevarlo a código de producción, ejecutar una monitorización ligera en un portátil o un gateway edge, o generar resúmenes periódicos para un informe. No sustituye a una base de datos de series temporales. Si necesitas meses de historial, varios usuarios consultando a la vez o almacenamiento duradero, escribe el flujo en InfluxDB o TimescaleDB y usa pandas sobre los resultados de las consultas.

Arquitectura: acumular, vaciar, analizar

El programa tiene dos hilos con tareas muy distintas.

1. El hilo de red de paho ejecuta on_message para cada mensaje entrante. Debe ser rápido, así que solo decodifica el JSON, extrae unos pocos campos y añade un diccionario simple a un collections.deque.
2. El hilo principal se despierta cada pocos segundos, vacía el deque, convierte las filas nuevas en un DataFrame, las añade a una ventana deslizante y ejecuta el análisis.

deque.append y deque.popleft son seguros entre hilos en CPython, así que no hace falta un lock explícito para este traspaso. Construir DataFrames dentro de on_message es el error más común en el código MQTT con pandas: cada construcción cuesta mucho más que añadir un dict y, mientras se ejecuta el callback, paho no puede procesar los keep-alives ni otros mensajes.

Prepara una fuente de datos

Necesitas un flujo con varios dispositivos y algo de ruido. La opción más sencilla es el simulador de nuestro tutorial del simulador de dispositivos MQTT, que publica JSON como este en connectmq/demo/<run-id>/<line>/<device>/telemetry en el servidor MQTT público gratuito:

{"device": "sensor-0042", "ts": 1760600000123, "temperature": 21.84, "humidity": 44.2}

Arráncalo, anota el run ID que imprime y úsalo más abajo. Cualquier otro flujo JSON también sirve, siempre que ajustes los nombres de los campos. Si eres nuevo interpretando payloads, la guía de manejo de JSON con MQTT en Python cubre la decodificación y la validación con más profundidad.

Instala las dependencias:

pip install "paho-mqtt>=2.0" "pandas>=2.1"

El script MQTT pandas completo

Guarda esto como mqtt_pandas.py y asigna a RUN_ID el run ID del simulador.

import json
import time
import uuid
from collections import deque

import pandas as pd
import paho.mqtt.client as mqtt

BROKER = "broker.connectmq.com"
PORT = 1883
RUN_ID = "3f9a1c2e"  # sustitúyelo por el run ID de tu simulador
TOPIC = f"connectmq/demo/{RUN_ID}/+/+/telemetry"
WINDOW = pd.Timedelta(minutes=15)  # cuánto historial mantener en memoria
ANALYSE_EVERY_S = 10
EXPECTED_INTERVAL_S = 30

inbox: deque = deque(maxlen=100_000)


def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        print(f"connect failed: {reason_code}")
        return
    client.subscribe(TOPIC, qos=0)
    print(f"subscribed to {TOPIC}")


def on_message(client, userdata, msg):
    try:
        data = json.loads(msg.payload)
    except (UnicodeDecodeError, json.JSONDecodeError):
        return  # ignorar payloads que no son JSON
    levels = msg.topic.split("/")
    inbox.append({
        "ts": data.get("ts", int(time.time() * 1000)),
        "line": levels[3],
        "device": levels[4],
        "temperature": data.get("temperature"),
        "humidity": data.get("humidity"),
    })


def drain() -> pd.DataFrame:
    rows = []
    while inbox:
        rows.append(inbox.popleft())
    if not rows:
        return pd.DataFrame()
    df = pd.DataFrame(rows)
    df["ts"] = pd.to_datetime(df["ts"], unit="ms", utc=True)
    df["temperature"] = pd.to_numeric(df["temperature"], errors="coerce")
    df["humidity"] = pd.to_numeric(df["humidity"], errors="coerce")
    return df


def analyse(window: pd.DataFrame) -> None:
    now = pd.Timestamp.now(tz="UTC")
    w = window.sort_values("ts")

    # 1. Media móvil de 5 minutos por dispositivo (ventana basada en tiempo)
    rolling = (
        w.groupby("device")
        .rolling("5min", on="ts")["temperature"]
        .mean()
        .groupby(level=0)
        .last()
        .rename("temp_5min_mean")
    )

    # 2. Remuestreo por línea y por minuto
    per_minute = (
        w.set_index("ts")
        .groupby("line")["temperature"]
        .resample("1min")
        .mean()
        .unstack(level=0)
        .round(2)
    )

    # 3. Señal robusta de anomalía: distancia a la mediana del dispositivo en unidades MAD
    med = w.groupby("device")["temperature"].transform("median")
    mad = (w["temperature"] - med).abs().groupby(w["device"]).transform("median")
    w = w.assign(score=(w["temperature"] - med).abs() / (mad * 1.4826 + 1e-9))
    anomalies = w[(w["score"] > 6) & (w["ts"] > now - pd.Timedelta(seconds=ANALYSE_EVERY_S * 2))]

    # 4. Dispositivos inactivos: sin datos durante más de dos intervalos
    last_seen = w.groupby("device")["ts"].max()
    stale = last_seen[last_seen < now - pd.Timedelta(seconds=EXPECTED_INTERVAL_S * 2)]

    print(f"\n=== {now:%H:%M:%S} | rows {len(w)} | devices {w['device'].nunique()} ===")
    print("Per-line means, last 3 minutes:")
    print(per_minute.tail(3).to_string())
    print("Hottest devices by 5-min mean:")
    print(rolling.sort_values(ascending=False).head(5).round(2).to_string())
    if not anomalies.empty:
        print("Anomalies:")
        print(anomalies[["ts", "device", "temperature"]].to_string(index=False))
    if not stale.empty:
        print(f"Stale devices: {', '.join(stale.index[:10])}")


def main():
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"pandas-{uuid.uuid4().hex[:8]}")
    client.on_connect = on_connect
    client.on_message = on_message
    client.connect(BROKER, PORT, keepalive=30)
    client.loop_start()

    window = pd.DataFrame()
    try:
        while True:
            time.sleep(ANALYSE_EVERY_S)
            new = drain()
            if not new.empty:
                window = pd.concat([window, new], ignore_index=True)
            if window.empty:
                continue
            cutoff = pd.Timestamp.now(tz="UTC") - WINDOW
            window = window[window["ts"] >= cutoff].reset_index(drop=True)
            analyse(window)
    except KeyboardInterrupt:
        window.to_csv("mqtt_snapshot.csv", index=False)
        print("saved mqtt_snapshot.csv")
    finally:
        client.loop_stop()
        client.disconnect()


if __name__ == "__main__":
    main()

Ejecútalo con python mqtt_pandas.py. Tras los primeros diez segundos verás una tabla de medias por línea, los cinco dispositivos más calientes y cualquier anomalía o sensor inactivo. Pulsa Ctrl+C para guardar la ventana actual como archivo CSV.

Repaso del análisis

Estadísticas móviles sobre una ventana temporal

rolling("5min", on="ts") usa una ventana basada en tiempo en lugar de un número fijo de filas. Esa distinción importa con datos IoT: los dispositivos publican a ritmos distintos y a veces se saltan lecturas, así que "las últimas 10 filas" pueden ser diez segundos para un dispositivo y diez minutos para otro. Una ventana temporal siempre cubre el mismo periodo. El rolling temporal exige que la columna de tiempo esté ordenada dentro de cada grupo, por eso la ventana se ordena antes por ts.

Remuestreo con pandas

resample("1min").mean() agrupa lecturas irregulares en intervalos regulares de un minuto por línea. El resultado es una tabla ancha con una columna por línea, lista para representar o exportar. Usa alias en minúsculas como "1min" y "10s"; las versiones recientes de pandas marcan como obsoleto el antiguo alias de segundos en mayúsculas. Cambia mean por agg(["mean", "min", "max", "count"]) si quieres un resumen más completo, y ten en cuenta count como indicador barato de calidad de datos.

Detección robusta de anomalías

Un z-score simple usa la media y la desviación típica, y ambas quedan distorsionadas precisamente por los picos que intentas encontrar. El script usa en su lugar la mediana y la desviación absoluta mediana (MAD), escalada por 1,4826 para que sea comparable a una desviación típica con datos normales. Una puntuación superior a seis es una señal fuerte de que una lectura es inverosímil. Los dispositivos defectuosos del simulador, que de vez en cuando saltan decenas de grados, se detectan al instante.

Detección de dispositivos inactivos

La comprobación de última aparición encuentra los dispositivos que llevan callados más de dos intervalos esperados. En producción suele ser más valiosa que cualquier umbral sobre los propios valores, porque un sensor muerto no informa de nada.

Consejos de rendimiento y memoria

Con unos cientos de mensajes por minuto, este script apenas usa CPU. Si la tasa de mensajes crece hasta miles por segundo, traslada el trabajo pesado a una base de datos o a un procesador de streams y reserva pandas para el análisis ad hoc.

Usar un broker privado para datos reales

El servidor público es perfecto para este tutorial, pero sus topics los puede leer cualquiera. Para datos reales de planta o de edificios, conéctate a un broker MQTT gestionado con credenciales y TLS. Solo cambian las líneas de conexión:

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="YOUR_CLIENT_ID")
client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD")
client.tls_set()  # certificado de Let's Encrypt, de confianza en el almacén del sistema
client.connect("broker.connectmq.com", 8883, keepalive=30)

Con ConnectMQ obtienes el usuario, la contraseña y el client ID en el portal tras crear un broker. Antes de ejecutar tu análisis, el explorador de topics en vivo del portal es una forma rápida de confirmar qué topics llegan realmente y qué aspecto tienen sus payloads.

Para conservar historial a largo plazo junto a tu vista en vivo con pandas, escribe el mismo flujo en una base de datos. Nuestra guía sobre cómo guardar datos MQTT en InfluxDB usa la misma estructura de topics, así que puedes ejecutar ambos en paralelo. Para otras opciones en Python, el resumen de bibliotecas cliente MQTT compara paho-mqtt con alternativas asíncronas, y la guía de usuario de pandas explica a fondo las ventanas y el remuestreo.

Empieza con ConnectMQ

Puedes ejecutar todo este tutorial contra broker.connectmq.com ahora mismo. Cuando tu análisis pase de datos de demostración a dispositivos reales, ConnectMQ te ofrece un broker privado con TLS, MQTT 5 y un portal para inspeccionar topics en vivo. Crea un broker Serverless gratuito en app.connectmq.com sin tarjeta de crédito. Si antes quieres entender las ventajas e inconvenientes del alojamiento, lee nuestra guía para elegir un broker MQTT en la nube.

Preguntas frecuentes

¿Puedo añadir cada mensaje MQTT directamente a un DataFrame de pandas?

Puedes, pero no deberías. Añadir fila a fila es lento y bloquea el hilo de red de paho, así que acumula diccionarios simples y construye un DataFrame con cada lote.

¿Cómo gestiono dispositivos que publican a ritmos distintos?

Usa ventanas basadas en tiempo como rolling("5min", on="ts") y resample("1min"). Cubren un periodo fijo independientemente de cuántas lecturas haya enviado cada dispositivo.

¿Qué marca de tiempo debo usar, la del payload o la de llegada?

Prefiere la marca de tiempo del dispositivo en el payload cuando su reloj sea fiable, porque sigue siendo correcta aunque los mensajes se retrasen. Recurre a la hora de llegada cuando el payload no la incluya.

¿Funciona este enfoque con clientes asyncio?

Sí. Con un cliente asíncrono el bucle de mensajes se ejecuta como una corrutina, pero el patrón es el mismo: recoge filas en un búfer y ejecuta pandas sobre lotes a intervalos fijos.