Jupyter Notebook'ta MQTT Verisini Keşfetmek: Subscribe, Grafik ve Anlık Görüntü Kaydı

Bir MQTT Jupyter notebook, daha önce hiç görmediğiniz bir veri akışını anlamanın en hızlı yoludur: hangi topic'ler var, payload'lar nasıl görünüyor ve değerler zaman içinde nasıl davranıyor. Bu rehberde bir notebook'tan broker'a bağlanacak, ham mesajları inceleyecek, canlı bir grafik çizecek, sonraki analizler için anlık görüntüler kaydedecek ve her şeyi düzgün şekilde kapatacaksınız.

MQTT Keşfi için Neden Jupyter Kullanmalı

Çoğu MQTT aracı bir seferde tek bir soruyu yanıtlar. Bir komut satırı subscriber'ı ham mesajları, bir masaüstü gezgini topic ağacını, bir panel ise grafikleri gösterir. Bir notebook bu üçünü tam Python veri yığınıyla birleştirir ve denediklerinizin kaydını tutar. Mevcut bir cihaz filosunu devraldığınızda, yeni bir sensör modeli aldığınızda veya garip bir değerin hatasını ayıklamanız gerektiğinde, bir notebook "ne publish ediliyor" sorusundan "işte bir grafik ve temizlenmiş bir veri seti" noktasına tek oturumda geçmenizi sağlar.

Kilit teknik ayrıntı, paho-mqtt'nin network loop'unu loop_start() ile bir arka plan thread'inde çalıştırabilmesidir. Mesajlar gelmeye devam ederken notebook kernel'i hücreleri çalıştırmak için serbest kalır. Bu rehberdeki her şey bunun üzerine kuruludur.

Kurulum

Python 3.9 veya daha yenisine ve bir Jupyter ortamına ihtiyacınız var: JupyterLab, klasik notebook arayüzü veya VS Code'daki notebook editörü bu rehber için aynı şekilde davranır. Paketleri kernel'in ortamına kurulsunlar diye notebook'un içinden kurun:

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

pyarrow yalnızca Parquet anlık görüntüleri için gereklidir. CSV'yi tercih ediyorsanız atlayabilirsiniz.

Hücre 1: Yapılandırma

Değiştirmek isteyebileceğiniz her şeyi ilk hücrede tutun. Rastgele bir run ID, herkesin her şeyi okuyabildiği herkese açık broker'da demo topic'lerinizi ayrı tutar.

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)

Zaten çalışan bir akışınız varsa, örneğin MQTT cihaz simülatörümüzden, PREFIX değerini bunun yerine o simülatörün önekine ayarlayın ve Hücre 3'ü atlayın.

Hücre 2: Subscriber'ı Arka Planda Bağlayın

# Bu hücreyi yeniden çalıştırmak geride eski bir istemci bırakmamalı
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()

En üstteki koruma, bir MQTT Jupyter notebook'taki en önemli satırdır. Notebook hücreleri sürekli yeniden çalıştırılır. Koruma olmadan her çalıştırma aynı client ID'ye sahip yeni bir istemci oluşturur, broker en yeni oturumu onurlandırmak için öncekinin bağlantısını keser, önceki istemcinin arka plan thread'i yeniden bağlanıp yenisini dışarı atar ve sonunda sonsuz bir döngüde çekişen iki istemciyle karşı karşıya kalırsınız. Mesajlar iki kez görünür ya da hiç görünmez ve nedeni açık değildir.

Callback'in ham baytları sakladığına ve hiçbir ayrıştırma yapmadığına dikkat edin. Callback'leri basit tutmak, hatalı bir payload'ın ağ thread'ini asla çökertemeyeceği ve nasıl çözeceğinize daha sonra karar verebileceğiniz anlamına gelir.

Hücre 3: Demo Veri Üretin (İsteğe Bağlı)

Keşfedecek bir akışınız yoksa bu hücre başka bir arka plan thread'inde küçük bir publisher başlatır. İki dakika boyunca üç sıcaklık sensörünü simüle eder ve herkese açık broker'ın istemci başına saniyede 10 mesaj sınırının oldukça altında kalır.

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()

Hücre 4: Ham Mesajları İnceleyin

Ayrıştırma kodu yazmadan önce gerçekte neyin geldiğine bakın. Gerçek dünyadaki payload'lar çoğu zaman dokümantasyonun vaat ettiği gibi değildir.

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)

value_counts() çıktısı size kaç farklı topic olduğunu ve her birinin ne kadar yoğun olduğunu söyler. Bu çoğu zaman bir sorunun ilk işaretidir; örneğin kardeşlerinden on kat daha sık publish yapan bir cihaz.

Hücre 5: Mesajları Bir DataFrame'e Dönüştürün

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()

Fonksiyon, cihaz adının dördüncü topic seviyesi olduğunu varsayar; bu da demo publisher ile eşleşir. Simülatörün topic'lerinde fazladan bir hat seviyesi vardır (<prefix>/<line>/<device>/telemetry); orada levels[4] kullanın ya da daha iyisi, payload'da varsa cihaz adını payload'dan okuyun. Bu tek satırı topic hiyerarşinize uyarlamak genellikle ihtiyacınız olan tek ayrıştırma değişikliğidir.

Cihaz başına describe(), sayıyı, ortalamayı, yayılımı ve aralığı tek bir tabloda verir. min veya max değerlerindeki aykırı değerler genellikle arızalı sensörlere veya Fahrenheit raporlayan bir cihaz gibi birim karışıklıklarına işaret eder.

Hücre 6: Notebook'ta Canlı Veriyi Çizin

Jupyter'daki en basit ve güvenilir canlı grafik, statik bir Matplotlib figürünü bir döngü içinde yeniden çizer ve önceki çıktıyı temizler. Ek widget'lar olmadan her notebook arayüzünde çalışır.

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)

Hücre, verdiğiniz süre boyunca bloklar; bu sırada arka plan thread'i mesaj toplamaya devam eder. Bittiğinde son grafik bir kayıt olarak notebook'ta kalır. Erken durdurmak için kernel'i kesin; subscriber kendi thread'inde yaşadığı için çalışmaya devam eder.

Hücre 7: Sonraki Analizler için Anlık Görüntü Kaydedin

Canlı veri kernel yeniden başladığında kaybolur. İlginç bir şey gördüğünüzde bir anlık görüntü kaydedin.

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")

Hem ayrıştırılmış tabloyu hem de ham mesajları kaydetmek bilinçli bir tercihtir. Daha sonra ayrıştırma mantığınızı değiştirdiğinizde, aynı durumun tekrar yaşanmasını beklemek yerine onu ham anlık görüntü üzerinde yeniden çalıştırabilirsiniz.

Hücre 8: Temizlik

sub.loop_stop()
sub.disconnect()

Notebook'u kapatmadan önce bunu çalıştırın. Kernel'in kapanması da thread'i sonlandırır, ancak açık bir bağlantı kesme, keep-alive süresi dolana kadar broker'da yarı açık bir oturum bırakmayı önler.

Sorun Giderme Kontrol Listesi

Notebook'tan Üretime

Notebook'lar keşif içindir. Hangi alanların önemli olduğunu öğrendiğinizde mantığı bir betiğe veya servise taşıyın: paho-mqtt Python rehberi bağımsız bir istemciyi ele alır, MQTT pandas analizi ise mikro toplu iş desenini uzun süre çalışan bir program olarak gösterir. Verinin saklanması gerekiyorsa, TIG stack kurulumunda anlatıldığı gibi bir zaman serisi veritabanı artı Grafana genellikle doğru adrestir.

Gerçek veriler için herkese açık broker'dan özel bir broker'a geçin. ConnectMQ gibi tam yönetilen bir MQTT hizmetinde yalnızca Hücre 2'deki bağlantı satırlarını değiştirirsiniz: sub.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD") çağırın, sub.tls_set() çağırın, client_id="YOUR_CLIENT_ID" kullanın ve 8883 portuna bağlanın. Broker'ın Let's Encrypt sertifikası varsayılan sistem deposu tarafından güvenilir kabul edilir. Bu adım için Python kütüphanelerini karşılaştırmak üzere MQTT istemci kütüphaneleri rehberine bakın; %pip beklenmedik davranırsa Jupyter dokümantasyonu kernel'leri ve ortamları ele alır.

ConnectMQ ile Başlayın

Bir notebook açın ve demo verileri hemen keşfetmek için onu herkese açık MQTT test broker'ına yönlendirin. Gerçek sensörlere geçtiğinizde ConnectMQ size TLS, MQTT 5 ve portalda notebook keşfiyle iyi eşleşen canlı bir topic gezgini içeren özel bir broker sunar. app.connectmq.com adresinde ücretsiz bir Serverless broker oluşturun, kredi kartı gerekmez.

Sık Sorulan Sorular

paho-mqtt bir Jupyter notebook içinde çalışır mı?

Evet. Network loop'un bir arka plan thread'inde çalışması için loop_start() kullanın. Bir notebook'ta loop_forever() kullanmaktan kaçının, çünkü siz kesene kadar hücreyi ve kernel'i bloklar.

Notebook'umda neden her MQTT mesajını iki kez alıyorum?

Büyük olasılıkla bağlantı hücresini yeniden çalıştırıp kendi aboneliğine sahip ikinci bir istemci oluşturdunuz ya da iki istemci aynı client ID için çekişiyor. Yenisini oluşturmadan önce eski istemciyi durdurun ve bağlantısını kesin.

Jupyter'da asyncio tabanlı bir MQTT istemcisi kullanabilir miyim?

Evet. Jupyter zaten bir event loop çalıştırır; bu yüzden bir hücrede asenkron bir istemciyi doğrudan await edebilirsiniz. Etkileşimli keşif için paho ile arka plan thread'i yaklaşımı daha basittir.

Anlık görüntüler için hangi formatı kullanmalıyım?

Parquet sütun türlerini ve zaman damgalarını korur ve kompakttır; bu da onu sonraki analizler için ideal kılar. CSV daha taşınabilirdir ancak tür bilgisini kaybeder.