Usar MQTT em um Jupyter Notebook é a forma mais rápida de entender um fluxo de dados que você nunca viu: quais topics existem, como são os payloads e como os valores se comportam ao longo do tempo. Neste guia você vai se conectar a um broker a partir de um notebook, inspecionar mensagens brutas, desenhar um gráfico ao vivo, salvar snapshots para análise posterior e encerrar tudo de forma limpa.
Por que usar o Jupyter para explorar MQTT
A maioria das ferramentas de MQTT responde a uma pergunta de cada vez. Um subscriber de linha de comando mostra as mensagens brutas, um explorador desktop mostra a árvore de topics e um dashboard mostra gráficos. Um notebook combina os três com toda a stack de dados do Python e ainda guarda um registro do que você tentou. Quando você herda uma frota de dispositivos existente, recebe um novo modelo de sensor ou precisa depurar um valor estranho, um notebook permite ir de "o que está sendo publicado" a "aqui está um gráfico e um conjunto de dados limpo" em uma única sessão.
O detalhe técnico principal é que o paho-mqtt consegue rodar o network loop em uma thread em segundo plano com loop_start(). O kernel do notebook fica livre para executar células enquanto as mensagens continuam chegando. Tudo neste guia se apoia nisso.
Preparação
Você precisa de Python 3.9 ou mais recente e de um ambiente Jupyter: JupyterLab, a interface clássica de notebook ou o editor de notebooks do VS Code se comportam da mesma forma para este guia. Instale os pacotes de dentro do notebook para que eles vão parar no ambiente do kernel:
%pip install "paho-mqtt>=2.0" pandas matplotlib pyarrow
O pyarrow só é necessário para snapshots em Parquet. Se preferir CSV, pode pulá-lo.
Célula 1: configuração
Deixe tudo o que você pode querer mudar na primeira célula. Um run ID aleatório mantém os seus topics de demonstração separados no servidor MQTT público, onde qualquer pessoa pode ler qualquer coisa.
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)
Se você já tem um fluxo rodando, por exemplo do nosso simulador de dispositivos MQTT, defina PREFIX com o prefixo desse simulador e pule a Célula 3.
Célula 2: conecte o subscriber em segundo plano
# Reexecutar esta célula não pode deixar um cliente antigo para trás
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()
A proteção no início é a linha mais importante de um notebook Jupyter com MQTT. As células de um notebook são reexecutadas o tempo todo. Sem essa proteção, cada execução cria mais um cliente com o mesmo client ID, o broker desconecta o anterior para respeitar a sessão mais nova, a thread em segundo plano do cliente anterior reconecta e derruba o novo, e você acaba com dois clientes brigando em um loop sem fim. As mensagens aparecem duas vezes, ou não aparecem, e a causa não é óbvia.
Repare que o callback guarda os bytes brutos e não faz parse nenhum. Manter os callbacks triviais garante que um payload malformado nunca derrube a thread de rede, e você decide depois como decodificar.
Célula 3: gere dados de demonstração (opcional)
Se você não tem um fluxo para explorar, esta célula inicia um pequeno publisher em outra thread em segundo plano. Ele simula três sensores de temperatura por dois minutos, bem abaixo do limite de 10 mensagens por segundo por cliente do servidor MQTT público.
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()
Célula 4: inspecione as mensagens brutas
Antes de escrever código de parse, olhe o que realmente está chegando. Payloads do mundo real frequentemente não são o que a documentação promete.
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)
A saída de value_counts() mostra quantos topics distintos existem e quão movimentado cada um está. Muitas vezes é o primeiro sinal de um problema, como um dispositivo publicando dez vezes mais do que os seus irmãos.
Célula 5: transforme as mensagens em um 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()
A função assume que o nome do dispositivo é o quarto nível do topic, o que corresponde ao publisher de demonstração. Os topics do simulador têm um nível extra de linha (<prefix>/<line>/<device>/telemetry), então use levels[4] nesse caso ou, melhor ainda, leia o nome do dispositivo do payload quando ele vier incluído. Adaptar essa única linha à sua hierarquia de topics costuma ser toda a mudança de parse de que você precisa.
O describe() por dispositivo mostra contagem, média, dispersão e intervalo em uma única tabela. Valores fora da curva em min ou max geralmente apontam para sensores defeituosos ou confusões de unidade, como um dispositivo reportando em Fahrenheit.
Célula 6: plote dados ao vivo no notebook
O gráfico ao vivo mais simples e confiável no Jupyter redesenha uma figura estática do Matplotlib em loop e limpa a saída anterior. Ele funciona em todas as interfaces de notebook, sem widgets extras.
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)
A célula fica bloqueada pela duração que você passar, enquanto a thread em segundo plano continua coletando mensagens. Ao terminar, o último gráfico permanece no notebook como registro. Interrompa o kernel para parar antes; o subscriber continua rodando, porque vive na própria thread.
Célula 7: salve snapshots para análise posterior
Dados ao vivo desaparecem quando o kernel reinicia. Salve um snapshot sempre que vir algo interessante.
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")
Salvar tanto o DataFrame processado quanto as mensagens brutas é proposital. Quando você mudar a lógica de parse mais tarde, pode reexecutá-la sobre o snapshot bruto, em vez de esperar a mesma situação acontecer de novo.
Célula 8: limpeza
sub.loop_stop()
sub.disconnect()
Execute isto antes de fechar o notebook. Desligar o kernel também encerra a thread, mas um disconnect explícito evita deixar uma sessão meio aberta no broker até o keep-alive expirar.
Checklist de solução de problemas
| Sintoma | Causa provável | Correção |
|---|---|---|
| Nenhuma mensagem chega | O topic assinado não corresponde ao prefixo do publisher | Imprima SUBSCRIBE_TO e compare com um subscriber de linha de comando |
| As mensagens chegam em rajadas e depois param | Dois clientes com o mesmo ID | Reexecute a Célula 2 com a proteção ou reinicie o kernel |
| O gráfico pisca ou se acumula | Falta o clear_output(wait=True) |
Mantenha a chamada de limpeza logo antes de plt.show() |
ModuleNotFoundError após a instalação |
Pacote instalado em outro ambiente | Use %pip dentro do notebook, não pip em um terminal |
| Timestamps em 1970 | O ts do payload está em segundos, não em milissegundos |
Troque unit="ms" por unit="s" |
| A memória do kernel não para de crescer | Armazenamento ilimitado | Mantenha maxlen no deque e salve snapshots em disco |
Do notebook para a produção
Notebooks servem para explorar. Quando você souber quais campos importam, leve a lógica para um script ou serviço: o guia de Paho MQTT em Python cobre um cliente independente, e a análise de MQTT com pandas mostra o padrão de microlotes como um programa de longa duração. Se os dados precisarem ser guardados, um banco de dados de séries temporais com Grafana costuma ser o lugar certo, como descrito na configuração da stack TIG.
Para dados reais, troque o servidor MQTT público por um broker privado. Em um broker MQTT gerenciado como o ConnectMQ, você só muda as linhas de conexão da Célula 2: chame sub.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD"), chame sub.tls_set(), use client_id="YOUR_CLIENT_ID" e conecte na porta 8883. O certificado Let's Encrypt do broker é confiável pelo repositório padrão do sistema. Para comparar bibliotecas Python para essa etapa, veja o guia de bibliotecas de cliente MQTT, e a documentação do Jupyter aborda kernels e ambientes caso o %pip se comporte de forma inesperada.
Comece a usar o ConnectMQ
Abra um notebook e aponte-o para o servidor MQTT público (gratuito), feito para testes, para explorar dados de demonstração agora mesmo. Quando passar para sensores reais, o ConnectMQ oferece um broker privado com TLS, MQTT 5 e um explorador de topics ao vivo no portal, que combina muito bem com a exploração em notebook. Crie um broker Serverless gratuito em app.connectmq.com, sem cartão de crédito.
Perguntas frequentes
O paho-mqtt funciona dentro de um Jupyter Notebook?
Sim. Use loop_start() para que o network loop rode em uma thread em segundo plano. Evite loop_forever() em um notebook, porque ele bloqueia a célula e o kernel até você interromper.
Por que recebo cada mensagem MQTT duas vezes no meu notebook?
O mais provável é que você tenha reexecutado a célula de conexão e criado um segundo cliente com a própria assinatura, ou que dois clientes estejam brigando pelo mesmo client ID. Pare e desconecte o cliente antigo antes de criar um novo.
Posso usar um cliente MQTT asyncio no Jupyter?
Sim. O Jupyter já roda um event loop, então você pode fazer await em um cliente assíncrono diretamente em uma célula. A abordagem com thread em segundo plano do paho é mais simples para exploração interativa.
Que formato devo usar para os snapshots?
O Parquet preserva os tipos das colunas e os timestamps e é compacto, o que o torna ideal para análise posterior. O CSV é mais portátil, mas perde as informações de tipo.
