MQTT-Daten mit Python in InfluxDB speichern

Um MQTT-Daten in InfluxDB zu speichern, brauchen Sie einen Subscriber, der jede Nachricht in einen Zeitreihenpunkt mit dem richtigen Measurement, den richtigen Tags, Fields und dem Zeitstempel umwandelt. In diesem Tutorial schreiben Sie diesen Subscriber in Python mit paho-mqtt und dem offiziellen InfluxDB-Client, leiten Tags automatisch aus Topic-Ebenen ab, schreiben in Batches und behandeln die Payload-Probleme, die echte Geräte erzeugen.

Warum eine Python-Bridge schreiben?

Telegraf kann MQTT abonnieren und allein per Konfiguration in InfluxDB schreiben, und für viele Teams ist das die richtige Wahl. Eine kurze Python-Bridge lohnt sich, wenn Sie Logik brauchen, die sich per Konfiguration nur umständlich ausdrücken lässt: Werte gegen physikalische Grenzen validieren, Einheiten umrechnen, Duplikate verwerfen, Punkte mit Metadaten aus einer Lookup-Tabelle anreichern oder mehrere Payload-Formate auf denselben Topics verarbeiten. Außerdem ist sie leicht zu lesen, zu testen und zu erweitern.

Die Bridge ist ein gewöhnlicher MQTT-Client. Sie verbindet sich mit dem Broker, abonniert und schreibt über dessen HTTP-API in InfluxDB. Es ist kein brokerseitiges Plugin und keine Integration beteiligt, sodass derselbe Code mit jedem standardkonformen Broker funktioniert, einschließlich des öffentlichen MQTT-Brokers unter broker.connectmq.com und eines privaten ConnectMQ-Brokers.

Datenmodell: von Topics zu Tags

InfluxDB organisiert Daten in Measurements, Tags und Fields. Tags sind indizierte Strings, nach denen Sie filtern und gruppieren. Fields sind die Werte, die Sie darstellen und aggregieren. Ein gut entworfenes MQTT-Topic enthält bereits die benötigten Tags:

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

Die Payload liefert die Fields und den Zeitstempel:

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

Der resultierende InfluxDB-Punkt sieht im Line Protocol so aus:

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

Topic-Ebenen nach Position abzubilden hält die Bridge generisch: Kommt eine neue Linie oder ein neues Gerät hinzu, ändert sich nichts am Code.

InfluxDB einrichten

Wenn Sie noch kein InfluxDB 2 haben, starten Sie es in Docker. Die Init-Variablen legen beim ersten Start eine Organisation, einen Bucket und ein Admin-Token an.

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

Für alles, was über einen lokalen Test hinausgeht, erstellen Sie in der InfluxDB-Oberfläche ein separates Token mit Schreibzugriff nur auf den Bucket sensors und verwenden dieses statt des Admin-Tokens.

Installieren Sie die Python-Pakete:

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

Die vollständige Bridge, um MQTT-Daten in InfluxDB zu speichern

Speichern Sie Folgendes als mqtt_influx_bridge.py. Die Konfiguration stammt aus Umgebungsvariablen, sodass dieselbe Datei für Tests und Produktion funktioniert.

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

# --- Konfiguration ---------------------------------------------------------
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")
# Ein Name pro Topic-Ebene: "_" ignoriert eine Ebene, "@measurement" benennt das 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"}  # Payload-Schlüssel, die keine Fields sind

# --- 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))  # Typkonflikte zwischen int und float vermeiden
        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:  # sieht nach Millisekunden aus
        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)  # gepuffert, nicht blockierend
    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()  # schreibt ausstehende Punkte
    influx.close()


if __name__ == "__main__":
    main()

Starten Sie sie und eine Datenquelle. Der MQTT-Gerätesimulator veröffentlicht genau dieses Format; setzen Sie MQTT_TOPIC auf sein Präfix:

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

Prüfen Sie die Daten anschließend im InfluxDB Data Explorer oder mit der 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

Designentscheidungen erklärt

Batching hält den Netzwerk-Thread schnell

write_api mit WriteOptions puffert Punkte und schreibt sie im Hintergrund, entweder sobald sich 500 Punkte angesammelt haben oder alle fünf Sekunden. Der on_message-Callback wartet nie auf eine HTTP-Anfrage, sodass paho weiterhin Keep-Alives und eingehende Nachrichten verarbeiten kann. Ein synchroner Schreibvorgang pro Nachricht funktioniert bei einer Handvoll Sensoren, gerät aber bei steigenden Raten schnell ins Hintertreffen. write_api.close() beim Herunterfahren schreibt alles, was noch gepuffert ist.

Zahlen in Float umwandeln vermeidet Typkonflikte

InfluxDB legt den Typ eines Fields beim ersten Schreiben in einen Shard fest. Sendet ein Gerät einmal "temperature": 21 und später 21.5, wird der zweite Schreibvorgang als Typkonflikt abgelehnt. Jede Zahl in Float umzuwandeln beseitigt diese ganze Fehlerklasse.

Tags müssen eine niedrige Kardinalität behalten

In Tags gehören nur stabile Bezeichner: Standort, Linie, Gerät, also die Ebenen der ISA-95-Hierarchie. Legen Sie niemals Zeitstempel, Sequenznummern oder Freitext in Tags, denn jede eindeutige Kombination erzeugt eine neue Serie, und die Anzahl der Serien bestimmt Speicherverbrauch und Abfragegeschwindigkeit.

Topic-Schema statt fest codierter Positionen

TOPIC_SCHEMA beschreibt die Hierarchie einmalig, sodass ein anderes Topic-Layout, etwa ein ISA-95-Pfad wie connectmq/factory1/production/line1/cnc1/telemetry, nur eine neue Umgebungsvariable benötigt: TOPIC_SCHEMA="_/site/area/line/device/@measurement".

Genau hier zahlt sich ein Unified Namespace (UNS) aus. Bildet die Topic-Struktur auf dem Broker die ISA-95-Equipment-Hierarchie ab (Enterprise, Site, Area, Line, Cell), etwa acme/werk-stuttgart/montage/linie-1/zelle-3/temperatur, dann wird jede Ebene des UNS direkt zu einem InfluxDB-Tag. Werk, Bereich und Linie lassen sich in Abfragen und Grafana-Dashboards filtern, ohne dass Sie eine eigene Zuordnungstabelle pflegen. Wie Sie diese Struktur entwerfen, erklärt der Beitrag zu Sparkplug B und Unified Namespace.

Andere Payload-Formen verarbeiten

Echte Flotten einigen sich selten auf ein Format. Zwei Varianten tauchen ständig auf, und beide passen mit wenigen Zeilen in to_point.

Reine Zahlen. Viele einfache Geräte veröffentlichen einen einzelnen Wert pro Topic, etwa 21.84 auf .../room1/temperature. Hier ist die letzte Topic-Ebene der Field-Name statt des Measurements. Versuchen Sie float(payload) vor dem JSON-Parsing, und wenn das gelingt, bauen Sie einen Punkt mit einem festen Measurement wie sensor und verwenden die letzte Topic-Ebene als Field-Schlüssel.

Verschachtelte Objekte. Industrielle Gateways senden oft Strukturen wie {"metrics": {"spindle_rpm": 11800, "temp_c": 41.7}}. Glätten Sie diese vor der Field-Schleife, indem Sie Schlüssel mit einem Unterstrich verbinden, sodass das Field metrics_spindle_rpm heißt. Begrenzen Sie die Glättungstiefe; tief verschachtelte Payloads enthalten meist Konfigurations- oder Diagnosedaten, die überhaupt nicht in eine Zeitreihendatenbank gehören.

Was auch immer Sie ergänzen, behalten Sie die Regel bei, dass unbekannte oder fehlerhafte Nachrichten gezählt und übersprungen werden und niemals eine Exception im Callback auslösen dürfen. Ein einziges fehlerhaftes Gerät darf die Bridge nicht für alle anderen stoppen.

Fehlerbehebung

Für eine tiefergehende Reconnect-Behandlung einschließlich Offline-Pufferung siehe robuste Reconnect-Logik mit paho MQTT. Wenn Sie stattdessen den reinen Konfigurationsweg bevorzugen, schreibt der Leitfaden zum Telegraf MQTT Consumer dasselbe Datenmodell ohne Code.

Checkliste für die Produktion

1. Verbinden Sie sich über TLS auf Port 8883 mit MQTT_TLS=1 und Zugangsdaten aus Ihrem Broker-Portal.
2. Verwenden Sie ein dediziertes InfluxDB-Token, das auf einen Bucket beschränkt ist.
3. Legen Sie für den Bucket eine Aufbewahrungsdauer fest, die dem entspricht, wie lange Sie Rohdaten tatsächlich brauchen.
4. Betreiben Sie die Bridge unter systemd oder als Container mit automatischem Neustart.
5. Abonnieren Sie mit QoS 1 und einer festen Client-ID, damit ein Neustart nicht stillschweigend Messwerte verwirft.
6. Überwachen Sie die Zähler written und skipped; ein plötzlicher Anstieg übersprungener Nachrichten bedeutet meist eine Firmware-Änderung.

Bei einem verwalteten MQTT-Broker wie ConnectMQ sind der erste und fünfte Punkt reine Konfiguration: Kopieren Sie YOUR_USERNAME, YOUR_PASSWORD und YOUR_CLIENT_ID aus dem Portal in die Umgebung. Die Bridge selbst ändert sich nicht. Beachten Sie, dass InfluxDB 3 einen anderen Python-Client verwendet; das Datenmodell aus diesem Leitfaden gilt weiterhin, die Schreibaufrufe unterscheiden sich jedoch. Die Dokumentation von influxdb-client behandelt den hier verwendeten 2.x-Client. Für Alternativen auf der Geräteseite siehe den Vergleich der MQTT-Client-Bibliotheken.

Starten Sie mit ConnectMQ

Testen Sie die Bridge zunächst mit simulierten Daten gegen den kostenlosen öffentlichen MQTT-Broker. Für echte Geräte bietet ConnectMQ einen privaten Broker mit TLS, MQTT 5 und einem Live-Topic-Explorer, der genau zeigt, welche Topics Ihre Bridge empfangen wird. Registrieren Sie sich unter app.connectmq.com für einen kostenlosen Serverless-Broker ohne Kreditkarte und lesen Sie mehr über das Modell des MQTT-Brokers in der Cloud, wenn Sie Wachstum planen.

Häufig gestellte Fragen

Sollte ich Telegraf oder Python verwenden, um MQTT-Daten in InfluxDB zu speichern?

Verwenden Sie Telegraf, wenn die Konfiguration Ihre Payloads abdeckt, da es keinen Code benötigt. Verwenden Sie Python, wenn Sie Validierung, Anreicherung, Einheitenumrechnung oder mehrere Payload-Formate brauchen.

Welche MQTT-Topic-Ebenen sollten zu Tags werden?

Stabile Bezeichner wie Standort, Linie und Gerät. Vermeiden Sie alles, was pro Nachricht eindeutig ist, denn eine hohe Tag-Kardinalität verlangsamt InfluxDB.

Warum werden einige meiner Punkte abgelehnt?

Der häufigste Grund ist ein Field-Typkonflikt, bei dem ein Field zuerst als Integer und später als Float oder String geschrieben wurde. Wandeln Sie Zahlen einheitlich in Float um.

Kann die Bridge Nicht-JSON-Payloads verarbeiten?

In der vorliegenden Form nicht; sie überspringt und zählt sie. Erweitern Sie to_point, um einfache Zahlen oder andere Formate zu parsen, wenn Ihre Geräte diese senden.