Log MQTT Messages to CSV with Python for Quick Analysis

When you need to log MQTT messages to CSV, you usually want answers fast: what did the sensor send overnight, how often does a device report, and does the data look sane before you build anything bigger. This tutorial builds a small Python subscriber that writes every message to CSV, rotates files by day and by size, and produces files that open cleanly in Excel, pandas or a Jupyter notebook. You can run it against the free public broker in five minutes and point it at a private broker later without changing the code.

Why CSV Is Still a Good First Step

A time-series database is the right destination for long-term data, but it is not always the right first step. CSV has a few properties that make it ideal for early analysis:

• Every tool reads it: spreadsheets, pandas, R, DuckDB, even a text editor.
• There is nothing to install or operate beyond Python.
• Files are easy to copy, email to a colleague or attach to a bug report.
• Append-only writes are simple and robust, so a crash loses at most the last line.

The trade-offs are just as clear. CSV has no indexes, no types and no concurrency control, so it gets slow and awkward past a few hundred megabytes, and it is a poor fit when several processes need to write. Treat it as a capture format for debugging, commissioning and short experiments. When the logger becomes permanent, graduate to something like MQTT to SQLite in Python or a proper time-series store.

What You Need

• Python 3.9 or newer
• paho-mqtt 2.x (pip install "paho-mqtt>=2.0")
• Optional: mosquitto_pub for sending test messages, and pandas for analysis

For testing, use the public MQTT test broker at broker.connectmq.com. It accepts anonymous connections with any non-empty, unique client ID. The public area is readable by everyone, so pick a random topic prefix such as connectmq/demo/csv-4821/ and never send real plant or customer data there.

Designing the CSV Layout

Before writing code, decide on the row shape. There are two common layouts:

The logger below uses the long layout because it never loses data: whatever arrives on the wire ends up in the file. It also tries to extract a numeric value column, either from a plain numeric payload like 21.4 or from a JSON object with a value key, which makes quick charts trivial without committing to a fixed schema.

Timestamps are written in UTC ISO 8601 with milliseconds. Local time looks friendlier, but it creates duplicate or missing hours around daylight-saving changes, and every analysis tool can convert UTC to local time on display.

The Python CSV Logger

Save this as mqtt_csv_logger.py:

import csv
import json
import os
import signal
import sys
import uuid
from datetime import datetime, timezone
from pathlib import Path

import paho.mqtt.client as mqtt

BROKER = os.environ.get("MQTT_HOST", "broker.connectmq.com")
PORT = int(os.environ.get("MQTT_PORT", "1883"))
USERNAME = os.environ.get("MQTT_USERNAME")          # None on the public broker
PASSWORD = os.environ.get("MQTT_PASSWORD")
CLIENT_ID = os.environ.get("MQTT_CLIENT_ID", f"csv-logger-{uuid.uuid4().hex[:8]}")
TOPICS = os.environ.get("MQTT_TOPICS", "connectmq/demo/csv-4821/#").split(",")
OUT_DIR = Path(os.environ.get("CSV_DIR", "mqtt_logs"))
MAX_BYTES = int(os.environ.get("CSV_MAX_BYTES", str(50 * 1024 * 1024)))
SKIP_RETAINED = os.environ.get("SKIP_RETAINED", "1") == "1"

FIELDS = ["received_at", "topic", "qos", "retain", "value", "payload"]


class RotatingCsvWriter:
    """Writes rows to mqtt_YYYY-MM-DD_NNN.csv, rotating daily and by size."""

    def __init__(self, directory: Path, max_bytes: int):
        self.directory = directory
        self.max_bytes = max_bytes
        self.directory.mkdir(parents=True, exist_ok=True)
        self._file = None
        self._writer = None
        self._day = None
        self._part = 0

    def _open(self, day: str):
        if self._file:
            self._file.close()
        if day != self._day:
            self._day, self._part = day, 0
        while True:
            path = self.directory / f"mqtt_{day}_{self._part:03d}.csv"
            if not path.exists() or path.stat().st_size < self.max_bytes:
                break
            self._part += 1
        new_file = not path.exists()
        # utf-8-sig adds a BOM so Excel detects UTF-8; newline="" is required by csv
        self._file = open(path, "a", newline="", encoding="utf-8-sig" if new_file else "utf-8")
        self._writer = csv.DictWriter(self._file, fieldnames=FIELDS)
        if new_file:
            self._writer.writeheader()
        print(f"Writing to {path}")

    def write(self, row: dict):
        day = row["received_at"][:10]
        if self._file is None or day != self._day or self._file.tell() >= self.max_bytes:
            if self._file is not None and day == self._day:
                self._part += 1
            self._open(day)
        self._writer.writerow(row)
        self._file.flush()

    def close(self):
        if self._file:
            self._file.close()


def extract_value(text: str):
    """Return a float from '21.4' or '{"value": 21.4}', else empty string."""
    try:
        return float(text)
    except ValueError:
        pass
    try:
        data = json.loads(text)
        if isinstance(data, dict) and isinstance(data.get("value"), (int, float)):
            return data["value"]
    except json.JSONDecodeError:
        pass
    return ""


writer = RotatingCsvWriter(OUT_DIR, MAX_BYTES)


def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        print(f"Connect failed: {reason_code}")
        return
    print(f"Connected as {CLIENT_ID}")
    for topic in TOPICS:
        client.subscribe(topic.strip(), qos=1)


def on_message(client, userdata, msg):
    if SKIP_RETAINED and msg.retain:
        return
    text = msg.payload.decode("utf-8", errors="replace")
    writer.write({
        "received_at": datetime.now(timezone.utc).isoformat(timespec="milliseconds"),
        "topic": msg.topic,
        "qos": msg.qos,
        "retain": int(msg.retain),
        "value": extract_value(text),
        "payload": text,
    })


def on_disconnect(client, userdata, flags, reason_code, properties):
    print(f"Disconnected: {reason_code}, paho will reconnect")


def main():
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID)
    if USERNAME:
        client.username_pw_set(USERNAME, PASSWORD)
    if PORT == 8883:
        client.tls_set()
    client.on_connect = on_connect
    client.on_message = on_message
    client.on_disconnect = on_disconnect
    client.reconnect_delay_set(min_delay=1, max_delay=30)

    def shutdown(*_):
        client.disconnect()
        writer.close()
        sys.exit(0)

    signal.signal(signal.SIGINT, shutdown)
    signal.signal(signal.SIGTERM, shutdown)

    client.connect(BROKER, PORT, keepalive=60)
    client.loop_forever(retry_first_connection=True)


if __name__ == "__main__":
    main()

A few design decisions are worth explaining:

• Single thread. loop_forever() runs callbacks on the same thread, so the writer needs no lock. If you switch to loop_start(), add a threading.Lock around write().
• Flush on every row. It costs a little throughput but means a power cut loses at most one line. For very high message rates, flush every N rows instead.
• Retained messages skipped by default. On every reconnect the broker replays retained messages, which would otherwise create duplicate rows each time the network blips.
• QoS 1 subscription. Messages published with QoS 1 are delivered at least once, so a reconnect may produce an occasional duplicate. That is usually fine for analysis; see MQTT QoS levels explained if you need to reason about it.

Run It and Send Test Data

Start the logger:

pip install "paho-mqtt>=2.0" pandas
python mqtt_csv_logger.py

In a second terminal, publish a few readings:

for i in 1 2 3 4 5; do
  mosquitto_pub -h broker.connectmq.com -p 1883 -i pub-csv-4821 \
    -t connectmq/demo/csv-4821/line1/temp -m "{\"value\": 2$i.5, \"unit\": \"C\"}"
  mosquitto_pub -h broker.connectmq.com -p 1883 -i pub-csv-4821 \
    -t connectmq/demo/csv-4821/line1/status -m "running"
  sleep 1
done

You should see a file like mqtt_logs/mqtt_2026-10-16_000.csv with a header and ten rows. Temperature rows have a number in value; status rows leave it empty but keep the raw text in payload. If you do not have a physical device, the MQTT device simulator article shows how to generate realistic streams.

Analyse the CSV with pandas

Long-format files pivot into charts with a few lines:

import glob
import pandas as pd

files = sorted(glob.glob("mqtt_logs/mqtt_*.csv"))
df = pd.concat((pd.read_csv(f, encoding="utf-8-sig") for f in files), ignore_index=True)
df["received_at"] = pd.to_datetime(df["received_at"], utc=True)

numeric = df.dropna(subset=["value"])
wide = numeric.pivot_table(index="received_at", columns="topic", values="value")
print(wide.resample("1min").mean().tail())

# Message rate per topic, useful for spotting chatty or silent devices
print(df.groupby("topic").size().sort_values(ascending=False))

The message-rate summary alone often justifies the logger: a device publishing ten times more than expected, or a topic that went silent at 03:00, shows up immediately.

Troubleshooting Checklist

1. Nothing is written. Check the topic filter. connectmq/demo/csv-4821/# matches everything below the prefix, but a typo in one level silently matches nothing. Subscribe with mosquitto_sub -v using the same filter to confirm traffic.
2. Excel shows everything in one column. Some regional settings expect semicolons. Use Data, From Text/CSV and choose comma as the delimiter, or set delimiter=";" in DictWriter for that audience.
3. Garbled characters in Excel. The BOM written by utf-8-sig fixes this for new files. Files created by older versions of the script need re-importing with UTF-8 selected.
4. Blank lines between rows on Windows. You opened the file without newline="". The script above already handles it.
5. The logger keeps disconnecting. Two processes are using the same client ID and kicking each other off. Let the script generate a random ID or set a unique MQTT_CLIENT_ID.
6. Binary payloads look like noise. Protobuf or compressed payloads are not text. Store msg.payload.hex() instead of the decoded text for those topics.

Moving to a Private Broker

The public broker is perfect for trying the logger, but real data belongs on a private broker with credentials. With ConnectMQ, you create a broker in the portal and receive a generated username, password and client ID. Then run the same script with environment variables and TLS on port 8883:

export MQTT_HOST=YOUR_BROKER_HOST
export MQTT_PORT=8883
export MQTT_USERNAME=YOUR_USERNAME
export MQTT_PASSWORD=YOUR_PASSWORD
export MQTT_CLIENT_ID=YOUR_CLIENT_ID
export MQTT_TOPICS="factory1/#"
python mqtt_csv_logger.py

Because client IDs on ConnectMQ are bound to the issued credentials, use the client ID from the portal rather than a random one. A managed MQTT broker also means the logger is just another client: you do not have to run Mosquitto next to it, and the portal's live topic explorer lets you check what the logger should be receiving before you trust the file.

When to Outgrow CSV

CSV logging is a stepping stone. Move on when any of these become true: files exceed what Excel opens comfortably, more than one process needs to write, you need queries across weeks of data, or other people need live dashboards. Natural next steps in this series are sending MQTT data to Google Sheets for shared lightweight reporting, and a real database for anything long-lived. For an overview of the Python, JavaScript and other clients you can use for these integrations, see the guide to MQTT client libraries. The paho-mqtt documentation covers every client option used here.

Get Started with ConnectMQ

Point your CSV logger at a private, TLS-secured broker in a few minutes. Sign up at app.connectmq.com, choose the free Serverless plan with no credit card, create a broker and paste the generated credentials into the environment variables above. You get up to 25 connections on the free plan, which is plenty for a logger and a handful of devices, plus an in-browser MQTT client for sending test messages.

Frequently Asked Questions

Can I log MQTT messages to CSV without writing code?

Yes, tools like Node-RED or Telegraf can write files, but a short Python script gives you full control over columns, rotation and retained-message handling. The script in this article is under 150 lines and has a single dependency.

Will I lose messages while the logger is offline?

With a clean session, yes: messages published while the logger is disconnected are not stored for it. Using a persistent session with QoS 1 lets the broker queue messages for a while, but for guaranteed history use a database fed by a long-running service.

How large can the CSV files get?

The script rotates at 50 MB by default and starts a new file every UTC day. Adjust CSV_MAX_BYTES to match the tool you analyse with; spreadsheets get sluggish long before pandas does.

Is it safe to log data from the public broker?

It is safe for test data. Anything on the public area of broker.connectmq.com can be read by anyone who subscribes to the same topic, so use a private broker with credentials for real readings.