Cómo guardar datos MQTT en InfluxDB con Python

Para guardar datos MQTT en InfluxDB necesitas un suscriptor que convierta cada mensaje en un punto de serie temporal con la measurement, los tags, los fields y la marca de tiempo correctos. En este tutorial escribirás ese suscriptor en Python con paho-mqtt y el cliente oficial de InfluxDB, derivarás los tags automáticamente de los niveles del topic, escribirás por lotes y gestionarás los problemas de payload que generan los dispositivos reales.

Por qué escribir un puente en Python

Telegraf puede suscribirse a MQTT y escribir en InfluxDB solo con configuración, y para muchos equipos es la opción correcta. Un puente corto en Python merece la pena cuando necesitas lógica que la configuración no expresa con comodidad: validar valores frente a límites físicos, convertir unidades, descartar duplicados, enriquecer puntos con metadatos de una tabla de consulta o manejar varios formatos de payload en los mismos topics. Además es fácil de leer, probar y ampliar.

El puente es un cliente MQTT corriente. Se conecta al broker, se suscribe y escribe en InfluxDB a través de su API HTTP. No interviene ningún plugin ni integración del lado del broker, lo que significa que el mismo código funciona con cualquier broker que cumpla el estándar, incluido el servidor MQTT público gratuito broker.connectmq.com y un broker privado de ConnectMQ.

Modelo de datos: de los topics a los tags

InfluxDB organiza los datos en measurements, tags y fields. Los tags son cadenas indexadas por las que filtras y agrupas. Los fields son los valores que representas y agregas. Un topic MQTT bien diseñado ya contiene los tags que necesitas:

connectmq/demo/a1b2c3/line1/sensor-0042/telemetry
   |       |     |      |        |          |
 ignore ignore ignore  line    device   measurement

El payload aporta los fields y la marca de tiempo:

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

El punto de InfluxDB resultante tiene este aspecto en line protocol:

telemetry,line=line1,device=sensor-0042 temperature=21.84,humidity=44.2 1760600000123000000

Asignar los niveles del topic por posición mantiene genérico el puente: si añades una nueva línea o un nuevo dispositivo, no cambia nada en el código.

Preparar InfluxDB

Si aún no tienes InfluxDB 2, ejecútalo en Docker. Las variables de inicialización crean una organización, un bucket y un token de administrador en el primer arranque.

docker run -d --name influxdb -p 8086:8086 \
  -v influxdb-data:/var/lib/influxdb2 \
  -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=iot \
  -e DOCKER_INFLUXDB_INIT_BUCKET=sensors \
  -e DOCKER_INFLUXDB_INIT_RETENTION=30d \
  -e DOCKER_INFLUXDB_INIT_ADMIN_TOKEN=my-super-secret-token \
  influxdb:2.7

Para cualquier cosa que vaya más allá de una prueba local, crea en la interfaz de InfluxDB un token separado con acceso de escritura solo al bucket sensors y úsalo en lugar del token de administrador.

Instala los paquetes de Python:

pip install "paho-mqtt>=2.0" influxdb-client

El puente completo para guardar datos MQTT en InfluxDB

Guarda lo siguiente como mqtt_influx_bridge.py. La configuración procede de variables de entorno, así que el mismo archivo sirve para pruebas y para producción.

import json
import logging
import os
import signal
import time
import uuid

import paho.mqtt.client as mqtt
from influxdb_client import InfluxDBClient, Point, WritePrecision, WriteOptions

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("bridge")

# --- configuración ---------------------------------------------------------
MQTT_HOST = os.getenv("MQTT_HOST", "broker.connectmq.com")
MQTT_PORT = int(os.getenv("MQTT_PORT", "1883"))
MQTT_TLS = os.getenv("MQTT_TLS", "0") == "1"
MQTT_USER = os.getenv("MQTT_USERNAME")
MQTT_PASS = os.getenv("MQTT_PASSWORD")
MQTT_CLIENT_ID = os.getenv("MQTT_CLIENT_ID", f"influx-bridge-{uuid.uuid4().hex[:8]}")
MQTT_TOPIC = os.getenv("MQTT_TOPIC", "connectmq/demo/a1b2c3/+/+/telemetry")
# Un nombre por nivel de topic: "_" ignora un nivel, "@measurement" nombra la measurement
TOPIC_SCHEMA = os.getenv("TOPIC_SCHEMA", "_/_/_/line/device/@measurement").split("/")

INFLUX_URL = os.getenv("INFLUX_URL", "http://localhost:8086")
INFLUX_TOKEN = os.environ["INFLUX_TOKEN"]
INFLUX_ORG = os.getenv("INFLUX_ORG", "iot")
INFLUX_BUCKET = os.getenv("INFLUX_BUCKET", "sensors")

SKIP_KEYS = {"ts", "device"}  # claves del payload que no son fields

# --- InfluxDB --------------------------------------------------------------
influx = InfluxDBClient(url=INFLUX_URL, token=INFLUX_TOKEN, org=INFLUX_ORG)
write_api = influx.write_api(write_options=WriteOptions(
    batch_size=500, flush_interval=5_000, retry_interval=5_000, max_retries=5))

stats = {"written": 0, "skipped": 0}


def to_point(topic: str, payload: bytes):
    levels = topic.split("/")
    if len(levels) != len(TOPIC_SCHEMA):
        return None
    try:
        data = json.loads(payload)
    except (UnicodeDecodeError, json.JSONDecodeError):
        return None
    if not isinstance(data, dict):
        return None

    measurement = "mqtt"
    tags = {}
    for name, value in zip(TOPIC_SCHEMA, levels):
        if name == "_":
            continue
        if name == "@measurement":
            measurement = value
        else:
            tags[name] = value

    point = Point(measurement)
    for k, v in tags.items():
        point = point.tag(k, v)

    fields = 0
    for key, value in data.items():
        if key in SKIP_KEYS:
            continue
        if isinstance(value, bool):
            point = point.field(key, value)
        elif isinstance(value, (int, float)):
            point = point.field(key, float(value))  # evita conflictos de tipo int/float
        elif isinstance(value, str) and len(value) <= 64:
            point = point.field(key, value)
        else:
            continue
        fields += 1
    if fields == 0:
        return None

    ts = data.get("ts")
    if isinstance(ts, (int, float)) and ts > 1e12:  # parece estar en milisegundos
        point = point.time(int(ts), WritePrecision.MS)
    else:
        point = point.time(time.time_ns(), WritePrecision.NS)
    return point


def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        log.error("MQTT connect failed: %s", reason_code)
        return
    client.subscribe(MQTT_TOPIC, qos=1)
    log.info("subscribed to %s", MQTT_TOPIC)


def on_message(client, userdata, msg):
    point = to_point(msg.topic, msg.payload)
    if point is None:
        stats["skipped"] += 1
        return
    write_api.write(bucket=INFLUX_BUCKET, record=point)  # con búfer, no bloqueante
    stats["written"] += 1


def main():
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=MQTT_CLIENT_ID)
    if MQTT_USER:
        client.username_pw_set(MQTT_USER, MQTT_PASS)
    if MQTT_TLS:
        client.tls_set()
    client.on_connect = on_connect
    client.on_message = on_message
    client.reconnect_delay_set(min_delay=1, max_delay=60)
    client.connect_async(MQTT_HOST, MQTT_PORT, keepalive=30)
    client.loop_start()

    running = True

    def stop(*_):
        nonlocal running
        running = False

    signal.signal(signal.SIGINT, stop)
    signal.signal(signal.SIGTERM, stop)

    while running:
        time.sleep(30)
        log.info("written=%d skipped=%d", stats["written"], stats["skipped"])

    client.loop_stop()
    client.disconnect()
    write_api.close()  # vuelca los puntos pendientes
    influx.close()


if __name__ == "__main__":
    main()

Ejecútalo y arranca una fuente de datos. El simulador de dispositivos MQTT publica exactamente este formato; asigna su prefijo a MQTT_TOPIC:

export INFLUX_TOKEN=my-super-secret-token
export MQTT_TOPIC="connectmq/demo/3f9a1c2e/+/+/telemetry"
python mqtt_influx_bridge.py

Después comprueba los datos en el Data Explorer de InfluxDB o con la CLI:

influx query 'from(bucket: "sensors")
  |> range(start: -10m)
  |> filter(fn: (r) => r._measurement == "telemetry" and r._field == "temperature")
  |> group(columns: ["line"])
  |> mean()' --org iot --token my-super-secret-token

Las decisiones de diseño, explicadas

Escribir por lotes mantiene rápido el hilo de red

write_api con WriteOptions acumula los puntos y los escribe en segundo plano, cuando se juntan 500 puntos o cada cinco segundos. El callback on_message nunca espera a una petición HTTP, así que paho puede seguir procesando keep-alives y mensajes entrantes. Una escritura síncrona por mensaje funciona con unos pocos sensores, pero se queda atrás rápidamente cuando aumentan las tasas. write_api.close() al apagar vuelca lo que quede en el búfer.

Convertir los números a float evita conflictos de tipo

InfluxDB fija el tipo de un field la primera vez que se escribe en un shard. Si un dispositivo envía una vez "temperature": 21 y más tarde 21.5, la segunda escritura se rechaza por conflicto de tipo. Convertir todos los números a float elimina toda esa clase de errores.

Los tags deben tener baja cardinalidad

Solo los identificadores estables pertenecen a los tags: emplazamiento, línea, dispositivo. No pongas nunca marcas de tiempo, números de secuencia ni mensajes de texto libre en los tags, porque cada combinación única crea una nueva serie y el número de series determina el uso de memoria y la velocidad de las consultas.

Esquema de topic en lugar de posiciones fijas en el código

TOPIC_SCHEMA describe la jerarquía una sola vez, así que una distribución de topics diferente, como una ruta ISA-95 del tipo connectmq/factory1/production/line1/cnc1/telemetry, solo necesita una nueva variable de entorno: TOPIC_SCHEMA="_/site/area/line/device/@measurement".

Manejar otras formas de payload

Las flotas reales rara vez se ponen de acuerdo en un único formato. Hay dos variantes que aparecen constantemente, y ambas encajan en to_point con unas pocas líneas.

Números sueltos. Muchos dispositivos sencillos publican un único valor por topic, como 21.84 en .../room1/temperature. Aquí el último nivel del topic es el nombre del field y no la measurement. Prueba float(payload) antes de interpretar el JSON y, si funciona, construye un punto con una measurement fija como sensor y usa el último nivel del topic como clave del field.

Objetos anidados. Los gateways industriales suelen enviar estructuras como {"metrics": {"spindle_rpm": 11800, "temp_c": 41.7}}. Aplánalas antes del bucle de fields, uniendo las claves con un guion bajo para que el field quede como metrics_spindle_rpm. Limita la profundidad del aplanado; los payloads muy anidados suelen contener datos de configuración o diagnóstico que ni siquiera pertenecen a una base de datos de series temporales.

Añadas lo que añadas, mantén la regla de que los mensajes desconocidos o mal formados se cuentan y se omiten, y nunca se les permite lanzar una excepción en el callback. Un solo dispositivo defectuoso no debe poder detener el puente para todos los demás.

Resolución de problemas

Para una gestión más profunda de las reconexiones, incluido el búfer offline, consulta la lógica robusta de reconexión de paho MQTT. Si prefieres la vía de solo configuración, la guía del Telegraf MQTT consumer escribe el mismo modelo de datos sin código.

Lista de comprobación para producción

1. Conéctate por TLS en el puerto 8883 con MQTT_TLS=1 y las credenciales del portal de tu broker.
2. Usa un token de InfluxDB dedicado y limitado a un bucket.
3. Fija en el bucket un periodo de retención acorde al tiempo que realmente necesitas los datos en bruto.
4. Ejecuta el puente con systemd o como contenedor con reinicio automático.
5. Suscríbete con QoS 1 y un client ID fijo para que un reinicio no descarte lecturas en silencio.
6. Vigila los contadores written y skipped; un aumento repentino de mensajes omitidos suele indicar un cambio de firmware.

Con un broker MQTT gestionado como ConnectMQ, los puntos primero y quinto son solo configuración: copia YOUR_USERNAME, YOUR_PASSWORD y YOUR_CLIENT_ID del portal al entorno. El puente en sí no cambia. Ten en cuenta que InfluxDB 3 usa un cliente de Python distinto; el modelo de datos de esta guía sigue siendo válido, pero las llamadas de escritura difieren. La documentación de influxdb-client cubre el cliente 2.x usado aquí. Para alternativas en el lado del dispositivo, consulta la comparativa de bibliotecas cliente MQTT.

Empieza con ConnectMQ

Prueba primero el puente contra el servidor MQTT público gratuito con datos simulados, nunca con datos de producción. Para dispositivos reales, ConnectMQ ofrece un broker privado con TLS, MQTT 5 y un explorador de topics en vivo que muestra exactamente qué topics recibirá tu puente. Regístrate en app.connectmq.com para obtener un broker Serverless gratuito sin tarjeta de crédito, y lee sobre el modelo de broker MQTT en la nube si planificas crecer.

Preguntas frecuentes

¿Debo usar Telegraf o Python para guardar datos MQTT en InfluxDB?

Usa Telegraf cuando la configuración cubra tus payloads, porque no requiere código. Usa Python cuando necesites validación, enriquecimiento, conversión de unidades o varios formatos de payload.

¿Qué niveles del topic MQTT deben convertirse en tags?

Identificadores estables como el emplazamiento, la línea y el dispositivo. Evita cualquier cosa única por mensaje, porque una alta cardinalidad de tags ralentiza InfluxDB.

¿Por qué se rechazan algunos de mis puntos?

La razón más común es un conflicto de tipo de field, cuando un field se escribió primero como entero y después como float o cadena. Convierte los números a float de forma coherente.

¿Puede el puente manejar payloads que no sean JSON?

Tal como está escrito, no; los omite y los cuenta. Amplía to_point para interpretar números sueltos u otros formatos si tus dispositivos los envían.