How to Store MQTT Data in InfluxDB with Python

To store MQTT data in InfluxDB you need a subscriber that turns each message into a time-series point with the right measurement, tags, fields and timestamp. In this tutorial you will write that subscriber in Python with paho-mqtt and the official InfluxDB client, derive tags automatically from topic levels, write in batches and handle the payload problems that real devices produce.

Why Write a Python Bridge

Telegraf can subscribe to MQTT and write to InfluxDB with configuration alone, and for many teams that is the right choice. A short Python bridge is worth it when you need logic that configuration cannot express comfortably: validating values against physical limits, converting units, dropping duplicates, enriching points with metadata from a lookup table, or handling several payload formats on the same topics. It is also easy to read, test and extend.

The bridge is an ordinary MQTT client. It connects to the broker, subscribes, and writes to InfluxDB through its HTTP API. No broker-side plugin or integration is involved, which means the same code works with any standards-compliant broker, including the public one at broker.connectmq.com and a private ConnectMQ broker.

Data Model: From Topics to Tags

InfluxDB organises data into measurements, tags and fields. Tags are indexed strings you filter and group by. Fields are the values you chart and aggregate. A well-designed MQTT topic already contains the tags you need:

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

The payload provides the fields and the timestamp:

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

The resulting InfluxDB point looks like this in line protocol:

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

Mapping topic levels by position keeps the bridge generic: if you add a new line or device, nothing in the code changes.

Set Up InfluxDB

If you do not already have InfluxDB 2, run it in Docker. The init variables create an organisation, a bucket and an admin token on first start.

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

For anything beyond a local test, create a separate token in the InfluxDB UI with write access to the sensors bucket only, and use that instead of the admin token.

Install the Python packages:

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

The Complete Bridge to Store MQTT Data in InfluxDB

Save the following as mqtt_influx_bridge.py. Configuration comes from environment variables so the same file works for testing and 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")
# One name per topic level: "_" ignores a level, "@measurement" names the 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 keys that are not 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))  # avoid int/float type conflicts
        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:  # looks like milliseconds
        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)  # buffered, non-blocking
    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()  # flushes pending points
    influx.close()


if __name__ == "__main__":
    main()

Run it and start a data source. The MQTT device simulator publishes exactly this format; set MQTT_TOPIC to its prefix:

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

Then check the data in the InfluxDB Data Explorer or with the 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

Design Decisions Explained

Batching keeps the network thread fast

write_api with WriteOptions buffers points and writes them in the background, either when 500 points accumulate or every five seconds. The on_message callback never waits for an HTTP request, so paho can keep processing keep-alives and incoming messages. A synchronous write per message works for a handful of sensors but falls behind quickly as rates grow. write_api.close() on shutdown flushes whatever is still buffered.

Casting numbers to float avoids type conflicts

InfluxDB fixes a field's type the first time it is written to a shard. If a device sends "temperature": 21 once and 21.5 later, the second write is rejected as a type conflict. Casting every number to float removes that whole class of error.

Tags must stay low-cardinality

Only stable identifiers belong in tags: site, line, device. Never put timestamps, sequence numbers or free-form messages in tags, because each unique combination creates a new series and series count drives memory use and query speed.

Topic schema instead of hard-coded positions

TOPIC_SCHEMA describes the hierarchy once, so a different topic layout, such as an ISA-95 path like connectmq/factory1/production/line1/cnc1/telemetry, only needs a new environment variable: TOPIC_SCHEMA="_/site/area/line/device/@measurement".

Handling Other Payload Shapes

Real fleets rarely agree on one format. Two variations come up constantly, and both fit into to_point with a few lines.

Bare numbers. Many simple devices publish a single value per topic, such as 21.84 on .../room1/temperature. Here the last topic level is the field name rather than the measurement. Try float(payload) before parsing JSON, and if it succeeds, build a point with a fixed measurement such as sensor and use the last topic level as the field key.

Nested objects. Industrial gateways often send structures like {"metrics": {"spindle_rpm": 11800, "temp_c": 41.7}}. Flatten them before the field loop, joining keys with an underscore so the field becomes metrics_spindle_rpm. Keep the flattening depth limited; deeply nested payloads usually contain configuration or diagnostic data that does not belong in a time-series database at all.

Whatever you add, keep the rule that unknown or malformed messages are counted and skipped, never allowed to raise an exception in the callback. One bad device should not be able to stop the bridge for everyone else.

Troubleshooting

For deeper reconnect handling, including offline buffering, see robust paho MQTT reconnect logic. If you want the configuration-only route instead, the Telegraf MQTT consumer guide writes the same data model without code.

Production Checklist

1. Connect over TLS on port 8883 with MQTT_TLS=1 and credentials from your broker portal.
2. Use a dedicated InfluxDB token scoped to one bucket.
3. Set a retention period on the bucket that matches how long you actually need raw data.
4. Run the bridge under systemd or as a container with automatic restart.
5. Subscribe with QoS 1 and a fixed client ID so a restart does not silently drop readings.
6. Monitor the written and skipped counters; a sudden rise in skipped messages usually means a firmware change.

With a fully managed MQTT service like ConnectMQ, the first and fifth items are configuration only: copy YOUR_USERNAME, YOUR_PASSWORD and YOUR_CLIENT_ID from the portal into the environment. The bridge itself does not change. Note that InfluxDB 3 uses a different Python client; the data model in this guide still applies, but the write calls differ. The influxdb-client documentation covers the 2.x client used here. For alternatives on the device side, see the MQTT client libraries comparison.

Get Started with ConnectMQ

Try the bridge against the free public MQTT broker with simulated data first. For real devices, ConnectMQ provides a private broker with TLS, MQTT 5 and a live topic explorer that shows exactly which topics your bridge will receive. Sign up at app.connectmq.com for a free Serverless broker with no credit card, and read about the MQTT broker in the cloud model if you are planning for growth.

Frequently Asked Questions

Should I use Telegraf or Python to store MQTT data in InfluxDB?

Use Telegraf when configuration covers your payloads, because it needs no code. Use Python when you need validation, enrichment, unit conversion or several payload formats.

Which MQTT topic levels should become tags?

Stable identifiers such as site, line and device. Avoid anything unique per message, because high tag cardinality slows InfluxDB down.

Why are some of my points rejected?

The most common reason is a field type conflict, where a field was first written as an integer and later as a float or string. Cast numbers to float consistently.

Can the bridge handle non-JSON payloads?

Not as written; it skips them and counts them. Extend to_point to parse plain numbers or other formats if your devices send them.