Telegraf MQTT Consumer: MQTT-Daten in InfluxDB übernehmen

Das Telegraf-Plugin MQTT Consumer ist der einfachste Weg, MQTT-Nachrichten ohne eigenen Code in eine Zeitreihendatenbank zu bringen. In diesem Tutorial konfigurieren Sie inputs.mqtt_consumer so, dass es Topics auf broker.connectmq.com abonniert, JSON-Payloads mit korrekten Zeitstempeln und Tags parst, das Ergebnis in InfluxDB 2 schreibt, und prüfen jeden Schritt.

Wie Telegraf in eine MQTT-Pipeline passt

Ein MQTT-Broker leitet Nachrichten weiter; er speichert keine Historie. Um die Temperaturen der letzten Woche darzustellen oder bei einem Trend zu alarmieren, muss etwas die Topics abonnieren und jeden Messwert in eine Datenbank schreiben. Telegraf ist ein kleiner Agent von InfluxData, der genau das tut. Er wird mit einer einzigen TOML-Datei konfiguriert, die Inputs, optionale Processors und Outputs enthält.

Der Input mqtt_consumer ist ein Service-Plugin: Er hält eine dauerhafte Verbindung zum Broker, empfängt Nachrichten bei ihrem Eintreffen, lässt sie durch einen Parser laufen, um Metriken zu erzeugen, und übergibt sie an Outputs, die stapelweise schreiben. Aus Sicht des Brokers ist Telegraf einfach ein weiterer MQTT-Client. Das ist auch das ehrliche Muster für jeden verwalteten MQTT-Broker, ConnectMQ eingeschlossen: Der Broker transportiert die Nachrichten, und ein externes Werkzeug wie Telegraf kümmert sich um die Speicherung.

Voraussetzungen

Wenn Sie noch kein InfluxDB haben, ist Docker der schnellste Weg:

docker run -d --name influxdb -p 8086:8086 \
  -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_ADMIN_TOKEN=my-super-secret-token \
  influxdb:2.7

Lieber ein einziger Befehl für den ganzen Stack? Der Leitfaden zum TIG-Stack mit Docker Compose startet Telegraf, InfluxDB und Grafana gemeinsam.

Topic- und Payload-Design

Dieser Leitfaden setzt die Topic-Struktur und JSON-Payload voraus, die in unseren Python-Tutorials verwendet werden:

connectmq/demo/<run-id>/<line>/<device>/telemetry

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

Wählen Sie Ihre eigene zufällige Run-ID, zum Beispiel a1b2c3, damit Ihre Topics auf dem öffentlichen MQTT-Broker nicht mit denen anderer kollidieren. Wenn Sie statt manueller Testnachrichten einen kontinuierlichen Datenstrom wollen, starten Sie den Python-MQTT-Gerätesimulator, der genau dieses Format veröffentlicht.

Eine vollständige Konfiguration für den Telegraf MQTT Consumer

Speichern Sie Folgendes als telegraf.conf:

[agent]
  interval = "10s"
  flush_interval = "10s"
  metric_batch_size = 1000
  metric_buffer_limit = 10000
  omit_hostname = true

[[inputs.mqtt_consumer]]
  servers = ["tcp://broker.connectmq.com:1883"]
  topics = ["connectmq/demo/a1b2c3/+/+/telemetry"]
  qos = 0
  client_id = "telegraf-a1b2c3"
  connection_timeout = "30s"

  # Name des Measurements in InfluxDB (Standard wäre "mqtt_consumer")
  name_override = "sensor"
  # Vollständiges Topic als Tag behalten; "" zum Deaktivieren
  topic_tag = "topic"

  # JSON-Payloads parsen
  data_format = "json"
  tag_keys = ["device"]
  json_time_key = "ts"
  json_time_format = "unix_ms"

  # Die Produktionslinie als zusätzliches Tag aus dem Topic extrahieren
  [[inputs.mqtt_consumer.topic_parsing]]
    topic = "connectmq/demo/a1b2c3/+/+/telemetry"
    tags = "_/_/_/line/_/_"

[[outputs.influxdb_v2]]
  urls = ["http://localhost:8086"]
  token = "${INFLUX_TOKEN}"
  organization = "iot"
  bucket = "sensors"

# Vorübergehend: Metriken beim Testen auch auf stdout ausgeben
[[outputs.file]]
  files = ["stdout"]
  data_format = "influx"

Was jede Einstellung bewirkt

• servers verwendet tcp:// für unverschlüsseltes MQTT auf Port 1883. Für TLS wechseln Sie zu ssl:// und Port 8883, wie unten gezeigt.
• topics akzeptiert MQTT-Wildcards. Die beiden +-Ebenen passen auf jede Linie und jedes Gerät, während das feste Suffix telemetry andere Nachrichtentypen ausschließt.
• client_id sollte explizit gesetzt werden. Ohne sie erzeugt Telegraf bei jedem Start eine zufällige ID, was persistente Sessions verhindert und den Client in Broker-Logs schwerer identifizierbar macht.
• name_override legt den Namen des InfluxDB-Measurements fest. Ohne diese Einstellung landet jede Metrik in einem Measurement namens mqtt_consumer.
• tag_keys macht JSON-Schlüssel zu Tags. Tags sind indiziert und eignen sich für Werte, nach denen Sie filtern oder gruppieren, etwa Geräte-IDs. Numerische Messwerte werden automatisch zu Fields.
• json_time_key und json_time_format verwenden den Zeitstempel des Geräts statt der Zeit, zu der Telegraf die Nachricht empfangen hat. unix_ms passt zu unseren Millisekunden-Zeitstempeln.
• topic_parsing ordnet Topic-Segmente Tags zu. Jedes _ überspringt eine Ebene, und benannte Segmente werden zu Tags, sodass line mit der vierten Ebene des Topics befüllt wird.

Ein Wort zum topic-Tag: Beim Debuggen ist es praktisch, doch jedes eindeutige Topic erzeugt in InfluxDB eine neue Serie. Bei einigen Hundert Geräten ist das harmlos. Bei Zehntausenden Topics oder Topics, die Nachrichten-IDs oder Zeitstempel enthalten, wächst die Serienkardinalität schnell und verlangsamt Abfragen. Sobald die Tags line und device dieselbe Information tragen, hält topic_tag = "" die Datenbank schlank.

Setzen Sie das Token als Umgebungsvariable und starten Sie Telegraf zuerst im Testmodus. Da mqtt_consumer ein Service-Input ist, benötigt der Testmodus eine Wartezeit, um Nachrichten zu sammeln:

export INFLUX_TOKEN=my-super-secret-token
telegraf --config telegraf.conf --test --test-wait 20

Veröffentlichen Sie währenddessen eine Testnachricht aus einem anderen Terminal:

mosquitto_pub -h broker.connectmq.com -p 1883 -i pub-a1b2c3 \
  -t connectmq/demo/a1b2c3/line1/sensor-0001/telemetry \
  -m "{\"device\":\"sensor-0001\",\"ts\":$(date +%s%3N),\"temperature\":21.7,\"humidity\":44.1}"

Der Aufruf date +%s%3N erzeugt mit GNU date unter Linux einen Millisekunden-Zeitstempel; unter macOS installieren Sie coreutils und verwenden gdate oder tippen einen festen aktuellen Wert ein. Telegraf sollte eine Zeile InfluxDB Line Protocol ähnlich dieser ausgeben:

sensor,device=sensor-0001,line=line1,topic=connectmq/demo/a1b2c3/line1/sensor-0001/telemetry humidity=44.1,temperature=21.7 1760600000123000000

Wenn das funktioniert, starten Sie Telegraf normal mit telegraf --config telegraf.conf, entfernen den outputs.file-Block, sobald Sie der Pipeline vertrauen, und fragen InfluxDB ab, um zu bestätigen, dass die Daten angekommen sind:

influx query 'from(bucket: "sensors") |> range(start: -15m) |> filter(fn: (r) => r._measurement == "sensor") |> limit(n: 5)' \
  --org iot --token my-super-secret-token

Die Verbindung für die Produktion absichern

Der öffentliche MQTT-Broker ist nur für Tests und Prototypen gedacht; alles darauf kann von anderen gelesen werden. Verwenden Sie für echte Daten einen privaten Broker mit Zugangsdaten und TLS. Bei einem verwalteten MQTT-Broker wie ConnectMQ ändert sich der Input-Block so:

[[inputs.mqtt_consumer]]
  servers = ["ssl://broker.connectmq.com:8883"]
  topics = ["connectmq/factory1/production/+/+/telemetry"]
  qos = 1
  client_id = "YOUR_CLIENT_ID"
  username = "${MQTT_USERNAME}"
  password = "${MQTT_PASSWORD}"
  persistent_session = true
  max_undelivered_messages = 1000

  name_override = "sensor"
  data_format = "json"
  tag_keys = ["device"]
  json_time_key = "ts"
  json_time_format = "unix_ms"

Der Broker präsentiert ein Let's-Encrypt-Zertifikat, dem der CA-Speicher des Betriebssystems vertraut, daher ist keine tls_ca-Einstellung nötig. persistent_session = true in Kombination mit QoS 1 und einer festen Client-ID weist den Broker an, die Session von Telegraf zu behalten und Nachrichten während kurzer Telegraf-Neustarts einzureihen. Bewahren Sie Zugangsdaten in Umgebungsvariablen statt in der Konfigurationsdatei auf, besonders wenn die Datei in der Versionsverwaltung liegt.

Der Topic-Filter im Beispiel folgt einer ISA-95-Struktur, wie sie ConnectMQ beim Republizieren von Sparkplug-B-Daten erzeugt. Das ist der Kern eines Unified Namespace (UNS): Die ISA-95-Equipment-Hierarchie (Enterprise → Site → Area → Line → Cell) wird zur Topic-Struktur auf dem MQTT-Broker, etwa acme/werk-stuttgart/montage/linie-1/zelle-3/temperatur. Telegraf abonniert dann einfach einen Zweig des UNS, und topic_parsing macht Werk, Linie und Maschine zu Tags in InfluxDB. Mehr dazu in Sparkplug B und Unified Namespace.

Fehlerbehebung beim Telegraf MQTT Consumer

Starten Sie Telegraf mit --debug, um Verbindungsereignisse und Parser-Fehler live zu sehen. Bei Payloads mit verschachtelten Objekten oder Arrays stößt der alte JSON-Parser schnell an seine Grenzen; der Leitfaden zum Telegraf-json_v2-Parser zeigt, wie Sie verschachtelte Strukturen und Topic-Segmente präzise abbilden. Die vollständige Optionsliste finden Sie in der offiziellen Dokumentation des mqtt_consumer-Plugins.

Wie es weitergeht

Sobald Daten in InfluxDB fließen, ergänzen Sie Grafana für Dashboards und Alarme. Wenn Sie Code der Konfiguration vorziehen, erledigt die Python-InfluxDB-Bridge dieselbe Aufgabe in einem kurzen Skript mit voller Kontrolle über das Mapping. Und wenn Sie noch Client-Werkzeuge für die Geräte selbst auswählen, lesen Sie den Leitfaden zu MQTT-Client-Bibliotheken.

Starten Sie mit ConnectMQ

Sie können diese gesamte Konfiguration in wenigen Minuten gegen den öffentlichen MQTT-Broker testen. Für Produktionsdaten bietet Ihnen ConnectMQ einen privaten Broker mit TLS auf Port 8883, MQTT 5 und einem Portal mit Live-Topic-Explorer, um zu prüfen, was Telegraf empfangen wird. Erstellen Sie unter app.connectmq.com einen kostenlosen Serverless-Broker ohne Kreditkarte und fügen Sie die generierten Zugangsdaten in Ihre Telegraf-Konfiguration ein. Wenn Sie ein größeres Deployment planen, lesen Sie mehr über den Betrieb eines Cloud-MQTT-Brokers.

Häufig gestellte Fragen

Unterstützt der Telegraf MQTT Consumer MQTT 5?

Neuere Telegraf-Versionen bieten eine protocol-Option, um MQTT 5 auszuwählen. Standard ist MQTT 3.1.1, was in den meisten Setups zum Abonnieren und Schreiben von Metriken ausreicht.

Kann eine Telegraf-Instanz mehrere Topics abonnieren?

Ja. Listen Sie mehrere Filter in topics auf oder fügen Sie mehrere mqtt_consumer-Blöcke mit unterschiedlichen Parsern hinzu, wenn Topics verschiedene Payload-Formate tragen.

Warum werden meine JSON-Strings nicht gespeichert?

Der alte json-Parser ignoriert String-Werte, sofern Sie sie nicht in json_string_fields oder tag_keys aufführen. Der json_v2-Parser behandelt Strings explizit.

Sollte Telegraf QoS 0 oder QoS 1 verwenden?

QoS 0 genügt für Telemetrie mit hoher Rate, bei der ein gelegentlich verlorener Messwert keine Rolle spielt. Verwenden Sie QoS 1 mit einer persistenten Session, wenn jeder Messwert gespeichert werden muss.