Exploring MQTT Data in a Jupyter Notebook: Subscribe, Plot and Save Snapshots

An MQTT Jupyter notebook is the quickest way to understand a data stream you have never seen before: what topics exist, what the payloads look like and how the values behave over time. In this guide you will connect to a broker from a notebook, inspect raw messages, draw a live chart, save snapshots for later analysis and shut everything down cleanly.

Why Use Jupyter for MQTT Exploration

Most MQTT tools answer one question at a time. A command-line subscriber shows raw messages, a desktop explorer shows the topic tree and a dashboard shows charts. A notebook combines all three with the full Python data stack, and it keeps a record of what you tried. When you inherit an existing device fleet, receive a new sensor model or need to debug a strange value, a notebook lets you move from "what is being published" to "here is a chart and a cleaned dataset" in one sitting.

The key technical detail is that paho-mqtt can run its network loop in a background thread with loop_start(). The notebook kernel stays free to execute cells while messages keep arriving. Everything in this guide builds on that.

Setup

You need Python 3.9 or newer and a Jupyter environment: JupyterLab, the classic notebook interface or the notebook editor in VS Code all behave the same way for this guide. Install the packages from inside the notebook so they land in the kernel's environment:

%pip install "paho-mqtt>=2.0" pandas matplotlib pyarrow

pyarrow is only needed for Parquet snapshots. If you prefer CSV, you can skip it.

Cell 1: Configuration

Keep everything you may want to change in the first cell. A random run ID keeps your demo topics separate on the public broker, where anyone can read anything.

import json, time, uuid, threading, random
from collections import deque
from datetime import datetime, timezone

import pandas as pd
import matplotlib.pyplot as plt
import paho.mqtt.client as mqtt

BROKER = "broker.connectmq.com"
PORT = 1883
RUN_ID = uuid.uuid4().hex[:8]
PREFIX = f"connectmq/demo/{RUN_ID}"
SUBSCRIBE_TO = f"{PREFIX}/#"
print("Topic prefix:", PREFIX)

If you already have a stream running, for example from our MQTT device simulator, set PREFIX to that simulator's prefix instead and skip Cell 3.

Cell 2: Connect the Subscriber in the Background

# Re-running this cell must not leave an old client behind
if "sub" in globals():
    sub.loop_stop()
    sub.disconnect()

messages = deque(maxlen=50_000)

def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        print("connect failed:", reason_code)
    else:
        client.subscribe(SUBSCRIBE_TO, qos=0)

def on_message(client, userdata, msg):
    messages.append({
        "received": datetime.now(timezone.utc),
        "topic": msg.topic,
        "payload": msg.payload,
        "retain": msg.retain,
    })

sub = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"nb-sub-{RUN_ID}")
sub.on_connect = on_connect
sub.on_message = on_message
sub.connect(BROKER, PORT, keepalive=30)
sub.loop_start()

The guard at the top is the single most important line in an MQTT Jupyter notebook. Notebook cells get re-run constantly. Without the guard, each run creates another client with the same client ID, the broker disconnects the previous one to honour the newest session, the previous client's background thread reconnects and kicks out the new one, and you end up with two clients fighting in an endless loop. Messages appear twice, or not at all, and the cause is not obvious.

Note that the callback stores raw bytes and does no parsing. Keeping callbacks trivial means a malformed payload can never crash the network thread, and you can decide how to decode later.

Cell 3: Generate Demo Data (Optional)

If you have no stream to explore, this cell starts a small publisher in another background thread. It simulates three temperature sensors for two minutes, staying well below the public broker's limit of 10 messages per second per client.

def demo_publisher(duration_s=120):
    pub = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"nb-pub-{RUN_ID}")
    pub.connect(BROKER, PORT, keepalive=30)
    pub.loop_start()
    state = {"room1": 21.0, "room2": 23.5, "server-rack": 30.0}
    end = time.time() + duration_s
    while time.time() < end:
        for device, value in state.items():
            value += random.gauss(0, 0.2)
            state[device] = value
            payload = {"ts": int(time.time() * 1000), "temperature": round(value, 2)}
            pub.publish(f"{PREFIX}/{device}/telemetry", json.dumps(payload))
        time.sleep(1)
    pub.loop_stop()
    pub.disconnect()

threading.Thread(target=demo_publisher, daemon=True).start()

Cell 4: Inspect Raw Messages

Before writing parsing code, look at what is really arriving. Real-world payloads are frequently not what the documentation promises.

print(len(messages), "messages so far")
for m in list(messages)[-5:]:
    print(m["topic"], m["payload"][:200])

topics = pd.Series([m["topic"] for m in messages]).value_counts()
topics.head(20)

The value_counts() output tells you how many distinct topics exist and how busy each one is. It is often the first sign of a problem, such as a device publishing ten times more often than its siblings.

Cell 5: Turn Messages into a DataFrame

def to_frame(msgs):
    rows = []
    for m in msgs:
        try:
            data = json.loads(m["payload"])
        except (UnicodeDecodeError, json.JSONDecodeError):
            continue
        if not isinstance(data, dict):
            continue
        levels = m["topic"].split("/")
        rows.append({
            "ts": pd.to_datetime(data.get("ts"), unit="ms", utc=True) if "ts" in data else m["received"],
            "device": levels[3] if len(levels) > 3 else m["topic"],
            "temperature": data.get("temperature"),
        })
    df = pd.DataFrame(rows)
    if not df.empty:
        df["temperature"] = pd.to_numeric(df["temperature"], errors="coerce")
        df = df.sort_values("ts")
    return df

df = to_frame(list(messages))
df.groupby("device")["temperature"].describe()

The function assumes the device name is the fourth topic level, which matches the demo publisher. The simulator's topics have an extra line level (<prefix>/<line>/<device>/telemetry), so use levels[4] there, or better, read the device name from the payload when it is included. Adapting this one line to your topic hierarchy is usually all the parsing change you need.

describe() per device gives count, mean, spread and range in one table. Outliers in min or max usually point to faulty sensors or unit mix-ups, such as one device reporting Fahrenheit.

Cell 6: Plot Live Data in the Notebook

The simplest reliable live chart in Jupyter redraws a static Matplotlib figure in a loop and clears the previous output. It works in every notebook front end without extra widgets.

from IPython.display import clear_output

def live_plot(seconds=60, every=2, window="5min"):
    end = time.time() + seconds
    while time.time() < end:
        df = to_frame(list(messages))
        if not df.empty:
            cutoff = df["ts"].max() - pd.Timedelta(window)
            recent = df[df["ts"] >= cutoff]
            fig, ax = plt.subplots(figsize=(10, 4))
            for device, g in recent.groupby("device"):
                ax.plot(g["ts"], g["temperature"], label=device)
            ax.set_title(f"Live temperature, last {window}")
            ax.set_ylabel("deg C")
            ax.legend(loc="upper left")
            clear_output(wait=True)
            plt.show()
        time.sleep(every)

live_plot(seconds=60)

The cell blocks for the duration you pass, while the background thread keeps collecting messages. When it finishes, the last chart stays in the notebook as a record. Interrupt the kernel to stop early; the subscriber keeps running because it lives in its own thread.

Cell 7: Save Snapshots for Later Analysis

Live data disappears when the kernel restarts. Save a snapshot whenever you see something interesting.

stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
df = to_frame(list(messages))
df.to_parquet(f"mqtt_snapshot_{stamp}.parquet", index=False)

raw = pd.DataFrame(list(messages))
raw["payload"] = raw["payload"].apply(lambda b: b.decode("utf-8", errors="replace"))
raw.to_parquet(f"mqtt_raw_{stamp}.parquet", index=False)
print("saved", len(df), "parsed rows and", len(raw), "raw messages")

Saving both the parsed frame and the raw messages is deliberate. When you later change your parsing logic, you can rerun it on the raw snapshot instead of waiting for the same situation to happen again.

Cell 8: Clean Up

sub.loop_stop()
sub.disconnect()

Run this before closing the notebook. A kernel shutdown also ends the thread, but an explicit disconnect avoids leaving a half-open session on the broker until the keep-alive expires.

Troubleshooting Checklist

From Notebook to Production

Notebooks are for exploration. Once you know which fields matter, move the logic into a script or service: the paho-mqtt Python guide covers a standalone client, and MQTT pandas analysis shows the micro-batch pattern as a long-running program. If the data needs to be kept, a time-series database plus Grafana is usually the right home, as described in the TIG stack setup.

For real data, switch from the public broker to a private one. On a fully managed MQTT service such as ConnectMQ, you only change the connection lines in Cell 2: call sub.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD"), call sub.tls_set(), use client_id="YOUR_CLIENT_ID" and connect to port 8883. The broker's Let's Encrypt certificate is trusted by the default system store. To compare Python libraries for that step, see the MQTT client libraries guide, and the Jupyter documentation covers kernels and environments if %pip behaves unexpectedly.

Get Started with ConnectMQ

Open a notebook and point it at the public MQTT test broker to explore demo data right now. When you move on to real sensors, ConnectMQ gives you a private broker with TLS, MQTT 5 and a live topic explorer in the portal, which pairs well with notebook exploration. Create a free Serverless broker at app.connectmq.com, no credit card needed.

Frequently Asked Questions

Does paho-mqtt work inside a Jupyter notebook?

Yes. Use loop_start() so the network loop runs in a background thread. Avoid loop_forever() in a notebook, because it blocks the cell and the kernel until you interrupt it.

Why do I receive every MQTT message twice in my notebook?

You most likely re-ran the connect cell and created a second client with its own subscription, or two clients are fighting over the same client ID. Stop and disconnect the old client before creating a new one.

Can I use an asyncio MQTT client in Jupyter?

Yes. Jupyter already runs an event loop, so you can await an async client directly in a cell. The background-thread approach with paho is simpler for interactive exploration.

What format should I use for snapshots?

Parquet keeps column types and timestamps intact and is compact, which makes it ideal for later analysis. CSV is more portable but loses type information.