Comment stocker des données MQTT dans InfluxDB avec Python

Pour stocker des données MQTT dans InfluxDB, il vous faut un subscriber qui transforme chaque message en point de série temporelle avec la bonne measurement, les bons tags, fields et horodatage. Dans ce tutoriel, vous allez écrire ce subscriber en Python avec paho-mqtt et le client InfluxDB officiel, dériver automatiquement les tags des niveaux de topic, écrire par lots et gérer les problèmes de payload que produisent les vrais appareils.

Pourquoi écrire une passerelle Python

Telegraf peut s'abonner à MQTT et écrire dans InfluxDB par simple configuration, et c'est le bon choix pour de nombreuses équipes. Une courte passerelle Python vaut la peine lorsque vous avez besoin d'une logique que la configuration exprime mal : valider des valeurs par rapport à des limites physiques, convertir des unités, éliminer des doublons, enrichir des points avec des métadonnées issues d'une table de correspondance, ou gérer plusieurs formats de payload sur les mêmes topics. Elle est aussi facile à lire, à tester et à étendre.

La passerelle est un client MQTT ordinaire. Elle se connecte au broker, s'abonne et écrit dans InfluxDB via son API HTTP. Aucun plugin ni intégration côté broker n'intervient : le même code fonctionne donc avec n'importe quel broker conforme aux standards, y compris le serveur MQTT public gratuit broker.connectmq.com et un broker ConnectMQ privé.

Modèle de données : des topics aux tags

InfluxDB organise les données en measurements, tags et fields. Les tags sont des chaînes indexées sur lesquelles vous filtrez et regroupez. Les fields sont les valeurs que vous tracez et agrégez. Un topic MQTT bien conçu contient déjà les tags dont vous avez besoin :

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

Le payload fournit les fields et l'horodatage :

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

Le point InfluxDB obtenu ressemble à ceci en line protocol :

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

Associer les niveaux de topic par position garde une passerelle générique : si vous ajoutez une nouvelle ligne ou un nouvel appareil, rien ne change dans le code.

Mettre en place InfluxDB

Si vous n'avez pas encore InfluxDB 2, lancez-le dans Docker. Les variables d'initialisation créent une organisation, un bucket et un token administrateur au premier démarrage.

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

Au-delà d'un test local, créez dans l'interface d'InfluxDB un token distinct avec un accès en écriture au seul bucket sensors, et utilisez-le à la place du token administrateur.

Installez les paquets Python :

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

La passerelle complète pour stocker des données MQTT dans InfluxDB

Enregistrez le code suivant sous mqtt_influx_bridge.py. La configuration provient de variables d'environnement, si bien que le même fichier sert pour les tests et pour la production.

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

# --- configuration ---------------------------------------------------------
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 nom par niveau de topic : "_" ignore un niveau, "@measurement" nomme 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"}  # clés du payload qui ne sont pas des 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))  # éviter les conflits de type 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:  # ressemble à des millisecondes
        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)  # mis en tampon, non bloquant
    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()  # écrit les points en attente
    influx.close()


if __name__ == "__main__":
    main()

Lancez-la et démarrez une source de données. Le simulateur d'appareils MQTT publie exactement ce format ; définissez MQTT_TOPIC avec son préfixe :

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

Vérifiez ensuite les données dans le Data Explorer d'InfluxDB ou avec 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

Les choix de conception expliqués

L'écriture par lots garde le thread réseau rapide

write_api avec WriteOptions met les points en tampon et les écrit en arrière-plan, soit dès que 500 points sont accumulés, soit toutes les cinq secondes. Le callback on_message n'attend jamais une requête HTTP, et paho peut ainsi continuer à traiter les keep-alives et les messages entrants. Une écriture synchrone par message convient pour une poignée de capteurs, mais prend vite du retard quand les débits augmentent. write_api.close() à l'arrêt écrit tout ce qui reste en tampon.

Convertir les nombres en float évite les conflits de type

InfluxDB fige le type d'un field lors de sa première écriture dans un shard. Si un appareil envoie une fois "temperature": 21 puis 21.5 plus tard, la seconde écriture est rejetée pour conflit de type. Convertir chaque nombre en float élimine toute cette catégorie d'erreurs.

Les tags doivent rester à faible cardinalité

Seuls des identifiants stables ont leur place dans les tags : site, ligne, appareil. Ne mettez jamais d'horodatages, de numéros de séquence ou de messages libres dans les tags, car chaque combinaison unique crée une nouvelle série, et le nombre de séries détermine la consommation mémoire et la vitesse des requêtes.

Un schéma de topic plutôt que des positions codées en dur

TOPIC_SCHEMA décrit la hiérarchie une seule fois : une autre organisation de topics, comme un chemin ISA-95 du type connectmq/factory1/production/line1/cnc1/telemetry, ne nécessite qu'une nouvelle variable d'environnement : TOPIC_SCHEMA="_/site/area/line/device/@measurement".

Gérer d'autres formes de payload

Les flottes réelles s'accordent rarement sur un format unique. Deux variantes reviennent sans cesse, et toutes deux s'intègrent dans to_point en quelques lignes.

Nombres bruts. De nombreux appareils simples publient une seule valeur par topic, comme 21.84 sur .../room1/temperature. Ici, le dernier niveau du topic est le nom du field plutôt que la measurement. Essayez float(payload) avant de parser le JSON et, en cas de succès, construisez un point avec une measurement fixe comme sensor en utilisant le dernier niveau du topic comme clé du field.

Objets imbriqués. Les passerelles industrielles envoient souvent des structures comme {"metrics": {"spindle_rpm": 11800, "temp_c": 41.7}}. Aplatissez-les avant la boucle des fields, en joignant les clés par un tiret bas pour que le field devienne metrics_spindle_rpm. Limitez la profondeur d'aplatissement ; les payloads profondément imbriqués contiennent généralement des données de configuration ou de diagnostic qui n'ont rien à faire dans une base de séries temporelles.

Quoi que vous ajoutiez, conservez la règle selon laquelle les messages inconnus ou malformés sont comptés et ignorés, sans jamais pouvoir lever d'exception dans le callback. Un seul appareil défaillant ne doit pas pouvoir arrêter la passerelle pour tous les autres.

Dépannage

Pour une gestion plus poussée de la reconnexion, avec mise en tampon hors ligne, consultez la logique de reconnexion paho MQTT robuste. Si vous préférez la voie purement configurable, le guide Telegraf MQTT consumer écrit le même modèle de données sans code.

Checklist de production

1. Connectez-vous en TLS sur le port 8883 avec MQTT_TLS=1 et les identifiants du portail de votre broker.
2. Utilisez un token InfluxDB dédié, limité à un seul bucket.
3. Définissez sur le bucket une durée de rétention correspondant au temps pendant lequel vous avez réellement besoin des données brutes.
4. Exécutez la passerelle sous systemd ou dans un conteneur avec redémarrage automatique.
5. Abonnez-vous en QoS 1 avec un client ID fixe pour qu'un redémarrage ne perde pas silencieusement de mesures.
6. Surveillez les compteurs written et skipped ; une hausse soudaine des messages ignorés signale généralement un changement de firmware.

Avec un broker MQTT managé comme ConnectMQ, le premier et le cinquième point se réduisent à de la configuration : copiez YOUR_USERNAME, YOUR_PASSWORD et YOUR_CLIENT_ID du portail dans l'environnement. La passerelle elle-même ne change pas. Notez qu'InfluxDB 3 utilise un client Python différent ; le modèle de données de ce guide reste valable, mais les appels d'écriture diffèrent. La documentation d'influxdb-client couvre le client 2.x utilisé ici. Pour les alternatives côté appareil, consultez le comparatif des bibliothèques clientes MQTT.

Démarrez avec ConnectMQ

Essayez d'abord la passerelle sur le serveur MQTT public gratuit avec des données simulées, jamais avec des données de production. Pour de vrais appareils, ConnectMQ fournit un broker privé avec TLS, MQTT 5 et un explorateur de topics en direct qui montre exactement quels topics votre passerelle recevra. Inscrivez-vous sur app.connectmq.com pour obtenir un broker Serverless gratuit sans carte bancaire, et découvrez le modèle du broker MQTT dans le cloud si vous anticipez une croissance.

Questions fréquentes

Faut-il utiliser Telegraf ou Python pour stocker des données MQTT dans InfluxDB ?

Utilisez Telegraf lorsque la configuration couvre vos payloads, car il ne nécessite aucun code. Utilisez Python lorsque vous avez besoin de validation, d'enrichissement, de conversion d'unités ou de plusieurs formats de payload.

Quels niveaux de topic MQTT doivent devenir des tags ?

Les identifiants stables comme le site, la ligne et l'appareil. Évitez tout ce qui est unique à chaque message, car une cardinalité de tags élevée ralentit InfluxDB.

Pourquoi certains de mes points sont-ils rejetés ?

La raison la plus fréquente est un conflit de type de field : un field écrit d'abord comme entier, puis comme float ou chaîne. Convertissez systématiquement les nombres en float.

La passerelle peut-elle gérer des payloads non JSON ?

Pas en l'état ; elle les ignore et les comptabilise. Étendez to_point pour parser des nombres simples ou d'autres formats si vos appareils en envoient.