Real-Time MQTT Data Analysis with Python and pandas

Combining MQTT and pandas gives you a fast way to analyse live IoT data in Python without standing up a database first. In this tutorial you will subscribe to a broker with paho-mqtt, buffer messages safely, turn them into DataFrames every few seconds and compute rolling statistics, per-minute resamples and simple anomaly flags.

When MQTT and pandas Are the Right Combination

pandas is a batch library. It is excellent at operating on a table of rows at once and poor at being updated one row at a time. MQTT is the opposite: messages arrive one by one, at irregular intervals, on a network thread you do not control. The trick is to put a small buffer between them and let pandas work on micro-batches.

This pattern is a good fit when you want to explore a new data stream, prototype a calculation before moving it into production code, run lightweight monitoring on a laptop or edge gateway, or produce periodic summaries for a report. It is not a replacement for a time-series database. If you need months of history, multiple users querying at once or durable storage, write the stream to InfluxDB or TimescaleDB and use pandas on query results instead.

Architecture: Buffer, Drain, Analyse

The program has two threads with very different jobs.

1. The paho network thread runs on_message for every incoming message. It must stay fast, so it only decodes JSON, extracts a few fields and appends a plain dictionary to a collections.deque.
2. The main thread wakes up every few seconds, drains the deque, converts the new rows to a DataFrame, appends them to a sliding window and runs the analysis.

deque.append and deque.popleft are thread safe in CPython, so no explicit lock is needed for this hand-off. Building DataFrames inside on_message is the most common mistake in MQTT pandas code: each construction costs far more than appending a dict, and while the callback runs, paho cannot process keep-alives or other messages.

Prepare a Data Source

You need a stream with several devices and some noise. The easiest option is the simulator from our MQTT device simulator tutorial, which publishes JSON like this to connectmq/demo/<run-id>/<line>/<device>/telemetry on the public broker:

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

Start it, note the run ID it prints, and use that ID below. Any other JSON stream works too, as long as you adjust the field names. If you are new to parsing payloads, the guide to Python MQTT JSON handling covers decoding and validation in more depth.

Install the dependencies:

pip install "paho-mqtt>=2.0" "pandas>=2.1"

The Complete MQTT pandas Script

Save this as mqtt_pandas.py and set RUN_ID to the simulator's run ID.

import json
import time
import uuid
from collections import deque

import pandas as pd
import paho.mqtt.client as mqtt

BROKER = "broker.connectmq.com"
PORT = 1883
RUN_ID = "3f9a1c2e"  # replace with your simulator run ID
TOPIC = f"connectmq/demo/{RUN_ID}/+/+/telemetry"
WINDOW = pd.Timedelta(minutes=15)  # how much history to keep in memory
ANALYSE_EVERY_S = 10
EXPECTED_INTERVAL_S = 30

inbox: deque = deque(maxlen=100_000)


def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        print(f"connect failed: {reason_code}")
        return
    client.subscribe(TOPIC, qos=0)
    print(f"subscribed to {TOPIC}")


def on_message(client, userdata, msg):
    try:
        data = json.loads(msg.payload)
    except (UnicodeDecodeError, json.JSONDecodeError):
        return  # ignore non-JSON payloads
    levels = msg.topic.split("/")
    inbox.append({
        "ts": data.get("ts", int(time.time() * 1000)),
        "line": levels[3],
        "device": levels[4],
        "temperature": data.get("temperature"),
        "humidity": data.get("humidity"),
    })


def drain() -> pd.DataFrame:
    rows = []
    while inbox:
        rows.append(inbox.popleft())
    if not rows:
        return pd.DataFrame()
    df = pd.DataFrame(rows)
    df["ts"] = pd.to_datetime(df["ts"], unit="ms", utc=True)
    df["temperature"] = pd.to_numeric(df["temperature"], errors="coerce")
    df["humidity"] = pd.to_numeric(df["humidity"], errors="coerce")
    return df


def analyse(window: pd.DataFrame) -> None:
    now = pd.Timestamp.now(tz="UTC")
    w = window.sort_values("ts")

    # 1. Rolling 5-minute mean per device (time-based window)
    rolling = (
        w.groupby("device")
        .rolling("5min", on="ts")["temperature"]
        .mean()
        .groupby(level=0)
        .last()
        .rename("temp_5min_mean")
    )

    # 2. Per-line, per-minute resample
    per_minute = (
        w.set_index("ts")
        .groupby("line")["temperature"]
        .resample("1min")
        .mean()
        .unstack(level=0)
        .round(2)
    )

    # 3. Robust anomaly flag: distance from the device median in MAD units
    med = w.groupby("device")["temperature"].transform("median")
    mad = (w["temperature"] - med).abs().groupby(w["device"]).transform("median")
    w = w.assign(score=(w["temperature"] - med).abs() / (mad * 1.4826 + 1e-9))
    anomalies = w[(w["score"] > 6) & (w["ts"] > now - pd.Timedelta(seconds=ANALYSE_EVERY_S * 2))]

    # 4. Stale devices: not seen for more than two intervals
    last_seen = w.groupby("device")["ts"].max()
    stale = last_seen[last_seen < now - pd.Timedelta(seconds=EXPECTED_INTERVAL_S * 2)]

    print(f"\n=== {now:%H:%M:%S} | rows {len(w)} | devices {w['device'].nunique()} ===")
    print("Per-line means, last 3 minutes:")
    print(per_minute.tail(3).to_string())
    print("Hottest devices by 5-min mean:")
    print(rolling.sort_values(ascending=False).head(5).round(2).to_string())
    if not anomalies.empty:
        print("Anomalies:")
        print(anomalies[["ts", "device", "temperature"]].to_string(index=False))
    if not stale.empty:
        print(f"Stale devices: {', '.join(stale.index[:10])}")


def main():
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"pandas-{uuid.uuid4().hex[:8]}")
    client.on_connect = on_connect
    client.on_message = on_message
    client.connect(BROKER, PORT, keepalive=30)
    client.loop_start()

    window = pd.DataFrame()
    try:
        while True:
            time.sleep(ANALYSE_EVERY_S)
            new = drain()
            if not new.empty:
                window = pd.concat([window, new], ignore_index=True)
            if window.empty:
                continue
            cutoff = pd.Timestamp.now(tz="UTC") - WINDOW
            window = window[window["ts"] >= cutoff].reset_index(drop=True)
            analyse(window)
    except KeyboardInterrupt:
        window.to_csv("mqtt_snapshot.csv", index=False)
        print("saved mqtt_snapshot.csv")
    finally:
        client.loop_stop()
        client.disconnect()


if __name__ == "__main__":
    main()

Run it with python mqtt_pandas.py. After the first ten seconds you will see a table of per-line averages, the five hottest devices and any anomalies or stale sensors. Press Ctrl+C to save the current window as a CSV file.

Walking Through the Analysis

Rolling statistics on a time-based window

rolling("5min", on="ts") uses a time-based window rather than a fixed number of rows. That distinction matters for IoT data: devices publish at different rates and sometimes skip readings, so "the last 10 rows" can mean ten seconds for one device and ten minutes for another. A time window always covers the same period. Time-based rolling requires the time column to be sorted within each group, which is why the window is sorted by ts first.

Resampling with pandas

resample("1min").mean() buckets irregular readings into regular one-minute intervals per line. The result is a wide table with one column per line, ready to plot or export. Use lowercase aliases such as "1min" and "10s"; recent pandas releases deprecate the older uppercase second alias. Swap mean for agg(["mean", "min", "max", "count"]) if you want a richer summary, and keep count in mind as a cheap data-quality indicator.

Robust anomaly detection

A plain z-score uses the mean and standard deviation, and both are distorted by the very spikes you are trying to find. The script instead uses the median and the median absolute deviation (MAD), scaled by 1.4826 so it is comparable to a standard deviation for normal data. A score above six is a strong signal that a reading is implausible. The simulator's faulty devices, which occasionally jump by tens of degrees, light up immediately.

Detecting stale devices

The last-seen check finds devices that have gone quiet for more than two expected intervals. In production this is often more valuable than any threshold on the values themselves, because a dead sensor reports nothing at all.

Performance and Memory Tips

At a few hundred messages per minute, this script uses very little CPU. If the message rate grows into thousands per second, move the heavy lifting to a database or a stream processor and keep pandas for ad hoc analysis.

Using a Private Broker for Real Data

The public broker is perfect for this tutorial, but its topics are readable by anyone. For real plant or building data, connect to a hosted MQTT broker with credentials and TLS. Only the connection lines change:

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="YOUR_CLIENT_ID")
client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD")
client.tls_set()  # Let's Encrypt certificate, trusted by the system store
client.connect("broker.connectmq.com", 8883, keepalive=30)

With ConnectMQ you get the username, password and client ID from the portal after creating a broker. Before running your analysis, the portal's live topic explorer is a quick way to confirm which topics are actually arriving and what their payloads look like.

To keep long-term history alongside your live pandas view, write the same stream to a database. Our guide on how to store MQTT data in InfluxDB uses the same topic structure, so you can run both side by side. For other Python options, the MQTT client libraries overview compares paho-mqtt with async alternatives, and the pandas user guide explains windowing and resampling in depth.

Get Started with ConnectMQ

You can run everything in this tutorial against broker.connectmq.com right now. When your analysis moves from demo data to real devices, ConnectMQ gives you a private broker with TLS, MQTT 5 and a portal to inspect live topics. Create a free Serverless broker at app.connectmq.com with no credit card required. If you want to understand the hosting trade-offs first, read our guide to choosing an MQTT broker in the cloud.

Frequently Asked Questions

Can I append each MQTT message directly to a pandas DataFrame?

You can, but you should not. Appending row by row is slow and blocks paho's network thread, so buffer plain dictionaries and build a DataFrame from each batch instead.

How do I handle devices that publish at different rates?

Use time-based windows such as rolling("5min", on="ts") and resample("1min"). They cover a fixed period regardless of how many readings each device sent.

Which timestamp should I use, the payload time or the arrival time?

Prefer the device timestamp in the payload when the device clock is trustworthy, because it stays correct even if messages are delayed. Fall back to arrival time when the payload has none.

Does this approach work with asyncio clients?

Yes. With an async client the message loop runs as a coroutine, but the pattern is the same: collect rows in a buffer and run pandas on batches at a fixed interval.