Collect data from MQTT

Subscribe to MQTT topics, turn topic paths into tags, parse the payload, and write the results to InfluxDB 3. This pattern fits IoT deployments where many devices publish readings to a structured topic tree.

This example assumes devices publish a bare numeric reading to topics shaped like:

sensors/<line>/<device>/temperature    payload: 76.2

Configuration

[[inputs.mqtt_consumer]]
  ## MQTT broker.
  servers = ["tcp://broker.example.com:1883"]

  ## Subscribe to all temperature readings in the topic tree.
  topics = ["sensors/+/+/temperature"]

  ## Deliver each message at least once and resume missed messages
  ## after a disconnect.
  qos = 1
  persistent_session = true
  client_id = "telegraf-plant-01"

  ## Credentials for the broker.
  username = "telegraf"
  password = "example-password"

  ## Parse the payload as a single float value.
  data_format = "value"
  data_type = "float"

  ## Extract the measurement name and tags from the topic path.
  [[inputs.mqtt_consumer.topic_parsing]]
    topic = "sensors/+/+/temperature"
    measurement = "_/_/_/measurement"
    tags = "_/line/device/_"

[[outputs.influxdb_v3]]
  urls = ["http://localhost:8181"]
  token = "
AUTH_TOKEN
"
  database = "
DATABASE_NAME
"

Replace the following:

• AUTH_TOKEN: your InfluxDB authorization token
• DATABASE_NAME: the database to write to

How it works

• topics uses MQTT wildcards: + matches one topic level and # matches any number of levels. This subscription receives temperature readings from every line and device.
• qos = 1 with persistent_session makes the broker hold messages published while Telegraf is offline and deliver them on reconnect. A stable client_id is required for the session to persist.
• The value parser reads the bare numeric payload as a float field named value.
• topic_parsing maps topic segments by position:
  • measurement = "_/_/_/measurement" uses the fourth segment (temperature) as the measurement name.
  • tags = "_/line/device/_" stores the second and third segments as the line and device tags.
  • _ marks segments to ignore. The full topic is also stored in the topic tag by default.
• Like all message-queue inputs, mqtt_consumer is a service input with delivery tracking: messages are acknowledged to the broker only after an output writes them.

Test the configuration

Service inputs deliver data only after messages arrive. Use --test-wait to keep the test running long enough to receive some:

telegraf --config mqtt.conf --test --test-wait 10

Example output

temperature,device=press-04,line=line-a,topic=sensors/line-a/press-04/temperature value=76.2 1709572232000000000
temperature,device=cnc-11,line=line-b,topic=sensors/line-b/cnc-11/temperature value=71.8 1709572233000000000

Parse JSON payloads

If your devices publish JSON instead of bare values, replace the parser configuration:

  data_format = "json_v2"
  [[inputs.mqtt_consumer.json_v2]]
    measurement_name = "sensors"
    [[inputs.mqtt_consumer.json_v2.field]]
      path = "temp"
    [[inputs.mqtt_consumer.json_v2.field]]
      path = "humidity"

Devices that report Unix-style timestamps in a local timezone can use the timestamp_tz formats. See Parse timestamps.

Extend this example

• For industrial equipment publishing Sparkplug B or OPC UA data, see Collect industrial data from OPC UA.
• To combine per-topic single values into one multi-field metric, use the merge aggregator or the pivot processor.