Registrar mensajes MQTT en CSV con Python

Cuando necesitas registrar mensajes MQTT en CSV, normalmente quieres respuestas rápidas: qué envió el sensor durante la noche, con qué frecuencia informa un dispositivo y si los datos parecen razonables antes de construir algo más grande. Este tutorial construye un pequeño suscriptor en Python que escribe cada mensaje en CSV, rota los archivos por día y por tamaño, y genera archivos que se abren sin problemas en Excel, pandas o un notebook de Jupyter. Puedes ejecutarlo contra el servidor MQTT público gratuito en cinco minutos y apuntarlo más adelante a un broker privado sin cambiar el código.

Por qué CSV sigue siendo un buen primer paso

Una base de datos de series temporales es el destino adecuado para los datos a largo plazo, pero no siempre es el primer paso adecuado. CSV tiene algunas propiedades que lo hacen ideal para un análisis inicial:

Todas las herramientas lo leen: hojas de cálculo, pandas, R, DuckDB e incluso un editor de texto.
No hay nada que instalar ni operar aparte de Python.
Los archivos son fáciles de copiar, enviar por correo a un compañero o adjuntar a un informe de errores.
Las escrituras de solo anexado son sencillas y robustas, así que un fallo pierde como mucho la última línea.

Las contrapartidas son igual de claras. CSV no tiene índices, ni tipos, ni control de concurrencia, así que se vuelve lento e incómodo a partir de unos cientos de megabytes, y encaja mal cuando varios procesos necesitan escribir. Trátalo como un formato de captura para depuración, puesta en marcha y experimentos breves. Cuando el registrador pase a ser permanente, da el salto a algo como MQTT a SQLite en Python o a un almacenamiento de series temporales en condiciones.

Qué necesitas

Python 3.9 o posterior
paho-mqtt 2.x (pip install "paho-mqtt>=2.0")
Opcional: mosquitto_pub para enviar mensajes de prueba y pandas para el análisis

Para las pruebas, usa ese servidor público en broker.connectmq.com, pensado para pruebas y prototipos. Acepta conexiones anónimas con cualquier client ID único y no vacío. El área pública es legible por todo el mundo, así que elige un prefijo de topic aleatorio como connectmq/demo/csv-4821/ y nunca envíes allí datos reales de planta o de clientes.

Diseñar la estructura del CSV

Antes de escribir código, decide la forma de las filas. Hay dos estructuras habituales:

Estructura	Columnas	Ideal para	Inconveniente
Larga (una fila por mensaje)	timestamp, topic, qos, retain, payload, value	Topics mezclados, payloads de forma desconocida, depuración	Necesita una tabla dinámica para los gráficos
Ancha (una columna por campo)	timestamp, device, temperature, humidity...	Un único esquema de payload conocido	Se rompe cuando aparece un campo nuevo

El registrador de abajo usa la estructura larga porque nunca pierde datos: todo lo que llega por la red acaba en el archivo. También intenta extraer una columna numérica value, ya sea de un payload numérico simple como 21.4 o de un objeto JSON con una clave value, lo que hace triviales los gráficos rápidos sin comprometerse con un esquema fijo.

Las marcas de tiempo se escriben en UTC en formato ISO 8601 con milisegundos. La hora local parece más amigable, pero genera horas duplicadas o ausentes en los cambios de horario de verano, y cualquier herramienta de análisis puede convertir UTC a hora local al mostrarla.

El registrador CSV en Python

Guarda esto como mqtt_csv_logger.py:

import csv
import json
import os
import signal
import sys
import uuid
from datetime import datetime, timezone
from pathlib import Path

import paho.mqtt.client as mqtt

BROKER = os.environ.get("MQTT_HOST", "broker.connectmq.com")
PORT = int(os.environ.get("MQTT_PORT", "1883"))
USERNAME = os.environ.get("MQTT_USERNAME")          # None en el broker público
PASSWORD = os.environ.get("MQTT_PASSWORD")
CLIENT_ID = os.environ.get("MQTT_CLIENT_ID", f"csv-logger-{uuid.uuid4().hex[:8]}")
TOPICS = os.environ.get("MQTT_TOPICS", "connectmq/demo/csv-4821/#").split(",")
OUT_DIR = Path(os.environ.get("CSV_DIR", "mqtt_logs"))
MAX_BYTES = int(os.environ.get("CSV_MAX_BYTES", str(50 * 1024 * 1024)))
SKIP_RETAINED = os.environ.get("SKIP_RETAINED", "1") == "1"

FIELDS = ["received_at", "topic", "qos", "retain", "value", "payload"]


class RotatingCsvWriter:
    """Escribe filas en mqtt_YYYY-MM-DD_NNN.csv, rotando a diario y por tamaño."""

    def __init__(self, directory: Path, max_bytes: int):
        self.directory = directory
        self.max_bytes = max_bytes
        self.directory.mkdir(parents=True, exist_ok=True)
        self._file = None
        self._writer = None
        self._day = None
        self._part = 0

    def _open(self, day: str):
        if self._file:
            self._file.close()
        if day != self._day:
            self._day, self._part = day, 0
        while True:
            path = self.directory / f"mqtt_{day}_{self._part:03d}.csv"
            if not path.exists() or path.stat().st_size < self.max_bytes:
                break
            self._part += 1
        new_file = not path.exists()
        # utf-8-sig añade un BOM para que Excel detecte UTF-8; csv requiere newline=""
        self._file = open(path, "a", newline="", encoding="utf-8-sig" if new_file else "utf-8")
        self._writer = csv.DictWriter(self._file, fieldnames=FIELDS)
        if new_file:
            self._writer.writeheader()
        print(f"Writing to {path}")

    def write(self, row: dict):
        day = row["received_at"][:10]
        if self._file is None or day != self._day or self._file.tell() >= self.max_bytes:
            if self._file is not None and day == self._day:
                self._part += 1
            self._open(day)
        self._writer.writerow(row)
        self._file.flush()

    def close(self):
        if self._file:
            self._file.close()


def extract_value(text: str):
    """Devuelve un float a partir de '21.4' o '{"value": 21.4}'; si no, cadena vacía."""
    try:
        return float(text)
    except ValueError:
        pass
    try:
        data = json.loads(text)
        if isinstance(data, dict) and isinstance(data.get("value"), (int, float)):
            return data["value"]
    except json.JSONDecodeError:
        pass
    return ""


writer = RotatingCsvWriter(OUT_DIR, MAX_BYTES)


def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        print(f"Connect failed: {reason_code}")
        return
    print(f"Connected as {CLIENT_ID}")
    for topic in TOPICS:
        client.subscribe(topic.strip(), qos=1)


def on_message(client, userdata, msg):
    if SKIP_RETAINED and msg.retain:
        return
    text = msg.payload.decode("utf-8", errors="replace")
    writer.write({
        "received_at": datetime.now(timezone.utc).isoformat(timespec="milliseconds"),
        "topic": msg.topic,
        "qos": msg.qos,
        "retain": int(msg.retain),
        "value": extract_value(text),
        "payload": text,
    })


def on_disconnect(client, userdata, flags, reason_code, properties):
    print(f"Disconnected: {reason_code}, paho will reconnect")


def main():
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID)
    if USERNAME:
        client.username_pw_set(USERNAME, PASSWORD)
    if PORT == 8883:
        client.tls_set()
    client.on_connect = on_connect
    client.on_message = on_message
    client.on_disconnect = on_disconnect
    client.reconnect_delay_set(min_delay=1, max_delay=30)

    def shutdown(*_):
        client.disconnect()
        writer.close()
        sys.exit(0)

    signal.signal(signal.SIGINT, shutdown)
    signal.signal(signal.SIGTERM, shutdown)

    client.connect(BROKER, PORT, keepalive=60)
    client.loop_forever(retry_first_connection=True)


if __name__ == "__main__":
    main()

Merece la pena explicar algunas decisiones de diseño:

Un solo hilo. loop_forever() ejecuta los callbacks en el mismo hilo, así que el escritor no necesita ningún lock. Si cambias a loop_start(), añade un threading.Lock alrededor de write().
Flush en cada fila. Cuesta un poco de rendimiento, pero significa que un corte de luz pierde como mucho una línea. Con tasas de mensajes muy altas, haz flush cada N filas.
Retained messages omitidos por defecto. En cada reconexión el broker vuelve a enviar los retained messages, lo que de otro modo crearía filas duplicadas cada vez que la red tiene un microcorte.
Suscripción con QoS 1. Los mensajes publicados con QoS 1 se entregan al menos una vez, así que una reconexión puede producir algún duplicado ocasional. Normalmente no es un problema para el análisis; consulta niveles de QoS en MQTT explicados si necesitas razonar sobre ello.

Ejecútalo y envía datos de prueba

Arranca el registrador:

pip install "paho-mqtt>=2.0" pandas
python mqtt_csv_logger.py

En una segunda terminal, publica algunas lecturas:

for i in 1 2 3 4 5; do
  mosquitto_pub -h broker.connectmq.com -p 1883 -i pub-csv-4821 \
    -t connectmq/demo/csv-4821/line1/temp -m "{\"value\": 2$i.5, \"unit\": \"C\"}"
  mosquitto_pub -h broker.connectmq.com -p 1883 -i pub-csv-4821 \
    -t connectmq/demo/csv-4821/line1/status -m "running"
  sleep 1
done

Deberías ver un archivo como mqtt_logs/mqtt_2026-10-16_000.csv con una cabecera y diez filas. Las filas de temperatura tienen un número en value; las de estado lo dejan vacío pero conservan el texto en bruto en payload. Si no tienes un dispositivo físico, el artículo sobre el simulador de dispositivos MQTT muestra cómo generar flujos realistas.

Analizar el CSV con pandas

Los archivos en formato largo se convierten en gráficos con unas pocas líneas:

import glob
import pandas as pd

files = sorted(glob.glob("mqtt_logs/mqtt_*.csv"))
df = pd.concat((pd.read_csv(f, encoding="utf-8-sig") for f in files), ignore_index=True)
df["received_at"] = pd.to_datetime(df["received_at"], utc=True)

numeric = df.dropna(subset=["value"])
wide = numeric.pivot_table(index="received_at", columns="topic", values="value")
print(wide.resample("1min").mean().tail())

# Tasa de mensajes por topic, útil para detectar dispositivos parlanchines o silenciosos
print(df.groupby("topic").size().sort_values(ascending=False))

El resumen de tasa de mensajes por sí solo suele justificar el registrador: un dispositivo que publica diez veces más de lo esperado, o un topic que enmudeció a las 03:00, salta a la vista de inmediato.

Lista de comprobación para solucionar problemas

No se escribe nada. Revisa el filtro de topic. connectmq/demo/csv-4821/# coincide con todo lo que hay bajo el prefijo, pero una errata en un nivel hace que no coincida con nada sin avisar. Suscríbete con mosquitto_sub -v usando el mismo filtro para confirmar que hay tráfico.
Excel muestra todo en una sola columna. Algunas configuraciones regionales esperan punto y coma. Usa Datos, Desde texto/CSV y elige la coma como delimitador, o define delimiter=";" en DictWriter para ese público.
Caracteres extraños en Excel. El BOM que escribe utf-8-sig lo soluciona para los archivos nuevos. Los archivos creados con versiones anteriores del script deben reimportarse seleccionando UTF-8.
Líneas en blanco entre filas en Windows. Abriste el archivo sin newline="". El script anterior ya lo tiene en cuenta.
El registrador se desconecta una y otra vez. Dos procesos usan el mismo client ID y se expulsan mutuamente. Deja que el script genere un ID aleatorio o define un MQTT_CLIENT_ID único.
Los payloads binarios parecen ruido. Los payloads Protobuf o comprimidos no son texto. Para esos topics, guarda msg.payload.hex() en lugar del texto decodificado.

Pasar a un broker privado

El broker público es perfecto para probar el registrador, pero los datos reales deben ir a un broker privado con credenciales. Con ConnectMQ, creas un broker en el portal y recibes un usuario, una contraseña y un client ID generados. Después ejecutas el mismo script con variables de entorno y TLS en el puerto 8883:

export MQTT_HOST=YOUR_BROKER_HOST
export MQTT_PORT=8883
export MQTT_USERNAME=YOUR_USERNAME
export MQTT_PASSWORD=YOUR_PASSWORD
export MQTT_CLIENT_ID=YOUR_CLIENT_ID
export MQTT_TOPICS="factory1/#"
python mqtt_csv_logger.py

Como en ConnectMQ los client ID están vinculados a las credenciales emitidas, usa el client ID del portal en lugar de uno aleatorio. Un broker MQTT gestionado también significa que el registrador es un cliente más: no tienes que ejecutar Mosquitto a su lado, y el explorador de topics en vivo del portal te permite comprobar lo que debería recibir el registrador antes de fiarte del archivo.

Cuándo se te queda pequeño el CSV

El registro en CSV es un paso intermedio. Pasa a otra cosa cuando se cumpla cualquiera de estas condiciones: los archivos superan lo que Excel abre con comodidad, más de un proceso necesita escribir, necesitas consultas que abarquen semanas de datos u otras personas necesitan dashboards en vivo. Los siguientes pasos naturales de esta serie son enviar datos MQTT a Google Sheets para informes ligeros compartidos y una base de datos real para todo lo que deba durar. Para una visión general de los clientes en Python, JavaScript y otros lenguajes que puedes usar en estas integraciones, consulta la guía de bibliotecas cliente MQTT. La documentación de paho-mqtt cubre todas las opciones de cliente usadas aquí.

Empieza con ConnectMQ

Apunta tu registrador CSV a un broker privado protegido con TLS en pocos minutos. Regístrate en app.connectmq.com, elige el plan gratuito Serverless sin tarjeta de crédito, crea un broker y pega las credenciales generadas en las variables de entorno anteriores. En el plan gratuito tienes hasta 25 conexiones, más que suficiente para un registrador y un puñado de dispositivos, además de un cliente MQTT en el navegador para enviar mensajes de prueba.

Preguntas frecuentes

¿Puedo registrar mensajes MQTT en CSV sin escribir código?

Sí, herramientas como Node-RED o Telegraf pueden escribir archivos, pero un script corto en Python te da control total sobre las columnas, la rotación y el tratamiento de los retained messages. El script de este artículo tiene menos de 150 líneas y una única dependencia.

¿Perderé mensajes mientras el registrador esté desconectado?

Con una sesión limpia, sí: los mensajes publicados mientras el registrador está desconectado no se guardan para él. Usar una sesión persistente con QoS 1 permite al broker encolar mensajes durante un tiempo, pero para tener un historial garantizado usa una base de datos alimentada por un servicio de larga duración.

¿Hasta qué tamaño pueden crecer los archivos CSV?

El script rota a los 50 MB por defecto y empieza un archivo nuevo cada día UTC. Ajusta CSV_MAX_BYTES a la herramienta con la que analizas; las hojas de cálculo se vuelven lentas mucho antes que pandas.

¿Es seguro registrar datos del broker público?

Es seguro para datos de prueba. Cualquiera que se suscriba al mismo topic puede leer todo lo que hay en el área pública de broker.connectmq.com, así que usa un broker privado con credenciales para las lecturas reales.

Registrar mensajes MQTT en CSV con Python para un análisis rápido