Python ve pandas ile Gerçek Zamanlı MQTT Veri Analizi

MQTT ile pandas'ı birleştirmek, önce bir veritabanı kurmadan Python'da canlı IoT verisini analiz etmenin hızlı bir yolunu sunar. Bu eğitimde paho-mqtt ile bir broker'a subscribe olacak, mesajları güvenli bir şekilde tamponlayacak, birkaç saniyede bir DataFrame'lere dönüştürecek ve kayan istatistikleri, dakikalık yeniden örneklemeleri ve basit anomali işaretlerini hesaplayacaksınız.

MQTT ve pandas Ne Zaman Doğru Birleşimdir

pandas bir toplu işlem (batch) kütüphanesidir. Bir satır tablosu üzerinde tek seferde işlem yapmakta mükemmeldir, satır satır güncellenmekte ise zayıftır. MQTT bunun tam tersidir: mesajlar kontrol etmediğiniz bir ağ thread'inde, düzensiz aralıklarla tek tek gelir. Püf noktası, aralarına küçük bir tampon koymak ve pandas'ın mikro toplu işler üzerinde çalışmasına izin vermektir.

Bu desen; yeni bir veri akışını keşfetmek, bir hesaplamayı üretim koduna taşımadan önce prototiplemek, bir dizüstü bilgisayarda veya edge gateway'de hafif izleme yapmak ya da bir rapor için periyodik özetler üretmek istediğinizde iyi bir seçimdir. Bir zaman serisi veritabanının yerine geçmez. Aylarca geçmişe, aynı anda sorgulama yapan birden fazla kullanıcıya veya kalıcı depolamaya ihtiyacınız varsa, akışı InfluxDB veya TimescaleDB'ye yazın ve pandas'ı bunun yerine sorgu sonuçları üzerinde kullanın.

Mimari: Tamponla, Boşalt, Analiz Et

Programın çok farklı işleri olan iki thread'i vardır.

1. paho ağ thread'i gelen her mesaj için on_message çalıştırır. Hızlı kalması gerekir; bu yüzden yalnızca JSON'u çözer, birkaç alanı çıkarır ve bir collections.deque içine düz bir sözlük ekler.
2. Ana thread birkaç saniyede bir uyanır, deque'yu boşaltır, yeni satırları bir DataFrame'e dönüştürür, bunları kayan bir pencereye ekler ve analizi çalıştırır.

deque.append ve deque.popleft CPython'da thread güvenlidir; bu yüzden bu devir için açık bir lock gerekmez. on_message içinde DataFrame oluşturmak, MQTT pandas kodundaki en yaygın hatadır: her oluşturma bir sözlük eklemekten çok daha pahalıdır ve callback çalışırken paho keep-alive'ları veya diğer mesajları işleyemez.

Bir Veri Kaynağı Hazırlayın

Birkaç cihaz ve biraz gürültü içeren bir akışa ihtiyacınız var. En kolay seçenek, herkese açık broker'da connectmq/demo/<run-id>/<line>/<device>/telemetry topic'ine bunun gibi JSON publish eden MQTT cihaz simülatörü eğitimimizdeki simülatördür:

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

Onu başlatın, yazdırdığı run ID'yi not edin ve aşağıda bu ID'yi kullanın. Alan adlarını uyarladığınız sürece başka herhangi bir JSON akışı da işe yarar. Payload ayrıştırmada yeniyseniz Python MQTT JSON işleme rehberi çözme ve doğrulamayı daha derinlemesine ele alır.

Bağımlılıkları kurun:

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

Eksiksiz MQTT pandas Betiği

Bunu mqtt_pandas.py olarak kaydedin ve RUN_ID değerini simülatörün run ID'si olarak ayarlayın.

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"  # simülatörünüzün run ID'siyle değiştirin
TOPIC = f"connectmq/demo/{RUN_ID}/+/+/telemetry"
WINDOW = pd.Timedelta(minutes=15)  # bellekte ne kadar geçmiş tutulacağı
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  # JSON olmayan payload'ları yok say
    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. Cihaz başına kayan 5 dakikalık ortalama (zamana dayalı pencere)
    rolling = (
        w.groupby("device")
        .rolling("5min", on="ts")["temperature"]
        .mean()
        .groupby(level=0)
        .last()
        .rename("temp_5min_mean")
    )

    # 2. Hat başına, dakika başına yeniden örnekleme
    per_minute = (
        w.set_index("ts")
        .groupby("line")["temperature"]
        .resample("1min")
        .mean()
        .unstack(level=0)
        .round(2)
    )

    # 3. Sağlam anomali işareti: cihaz medyanından MAD birimi cinsinden uzaklık
    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. Bayat cihazlar: iki aralıktan uzun süredir görülmeyenler
    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()

python mqtt_pandas.py ile çalıştırın. İlk on saniyeden sonra hat başına ortalamaların, en sıcak beş cihazın ve varsa anomalilerin veya bayat sensörlerin bir tablosunu görürsünüz. Mevcut pencereyi bir CSV dosyası olarak kaydetmek için Ctrl+C'ye basın.

Analizi Adım Adım İnceleme

Zamana dayalı pencerede kayan istatistikler

rolling("5min", on="ts") sabit sayıda satır yerine zamana dayalı bir pencere kullanır. Bu ayrım IoT verisi için önemlidir: cihazlar farklı hızlarda publish eder ve bazen ölçümleri atlar; bu yüzden "son 10 satır" bir cihaz için on saniye, bir diğeri için on dakika anlamına gelebilir. Bir zaman penceresi her zaman aynı süreyi kapsar. Zamana dayalı kayan pencere, zaman sütununun her grup içinde sıralı olmasını gerektirir; pencerenin önce ts değerine göre sıralanmasının nedeni budur.

pandas ile yeniden örnekleme

resample("1min").mean(), düzensiz ölçümleri hat başına düzenli bir dakikalık aralıklara gruplar. Sonuç, hat başına bir sütun içeren, grafiğe dökülmeye veya dışa aktarılmaya hazır geniş bir tablodur. "1min" ve "10s" gibi küçük harfli takma adlar kullanın; güncel pandas sürümleri eski büyük harfli saniye takma adını kullanımdan kaldırıyor. Daha zengin bir özet istiyorsanız mean yerine agg(["mean", "min", "max", "count"]) kullanın ve count değerini ucuz bir veri kalitesi göstergesi olarak aklınızda tutun.

Sağlam anomali tespiti

Düz bir z-skoru ortalamayı ve standart sapmayı kullanır; ikisi de bulmaya çalıştığınız sıçramalar tarafından bozulur. Betik bunun yerine medyanı ve normal veri için bir standart sapmayla karşılaştırılabilir olması amacıyla 1,4826 ile ölçeklenen medyan mutlak sapmayı (MAD) kullanır. Altının üzerindeki bir skor, bir ölçümün mantıksız olduğuna dair güçlü bir işarettir. Simülatörün ara sıra onlarca derece sıçrayan arızalı cihazları hemen göze çarpar.

Bayat cihazları tespit etmek

Son görülme kontrolü, beklenen iki aralıktan daha uzun süredir sessiz kalan cihazları bulur. Üretimde bu, çoğu zaman değerlerin kendisine uygulanan herhangi bir eşikten daha değerlidir, çünkü ölü bir sensör hiçbir şey raporlamaz.

Performans ve Bellek İpuçları

Dakikada birkaç yüz mesajda bu betik çok az CPU kullanır. Mesaj hızı saniyede binlere çıkarsa ağır işi bir veritabanına veya bir akış işlemcisine taşıyın ve pandas'ı anlık analizler için tutun.

Gerçek Veriler için Özel Bir Broker Kullanmak

Herkese açık broker bu eğitim için mükemmeldir, ancak topic'lerini herkes okuyabilir. Gerçek tesis veya bina verileri için kimlik bilgileri ve TLS ile barındırılan bir MQTT broker'ına bağlanın. Yalnızca bağlantı satırları değişir:

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="YOUR_CLIENT_ID")
client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD")
client.tls_set()  # Sistem deposunun güvendiği Let's Encrypt sertifikası
client.connect("broker.connectmq.com", 8883, keepalive=30)

ConnectMQ ile bir broker oluşturduktan sonra kullanıcı adını, parolayı ve client ID'yi portaldan alırsınız. Analizinizi çalıştırmadan önce portalın canlı topic gezgini, hangi topic'lerin gerçekten geldiğini ve payload'larının nasıl göründüğünü doğrulamanın hızlı bir yoludur.

Canlı pandas görünümünüzün yanında uzun vadeli geçmiş tutmak için aynı akışı bir veritabanına yazın. MQTT verisini InfluxDB'de saklama rehberimiz aynı topic yapısını kullanır, böylece ikisini yan yana çalıştırabilirsiniz. Diğer Python seçenekleri için MQTT istemci kütüphanelerine genel bakış paho-mqtt'yi asenkron alternatiflerle karşılaştırır; pandas kullanıcı kılavuzu ise pencereleme ve yeniden örneklemeyi derinlemesine açıklar.

ConnectMQ ile Başlayın

Bu eğitimdeki her şeyi şu anda broker.connectmq.com üzerinde çalıştırabilirsiniz. Analiziniz demo verilerden gerçek cihazlara geçtiğinde ConnectMQ size TLS, MQTT 5 ve canlı topic'leri incelemek için bir portal içeren özel bir broker sunar. Kredi kartı gerektirmeden app.connectmq.com adresinde ücretsiz bir Serverless broker oluşturun. Önce barındırma ödünleşimlerini anlamak istiyorsanız, bulutta MQTT broker seçimi rehberimizi okuyun.

Sık Sorulan Sorular

Her MQTT mesajını doğrudan bir pandas DataFrame'ine ekleyebilir miyim?

Ekleyebilirsiniz, ama eklememelisiniz. Satır satır eklemek yavaştır ve paho'nun ağ thread'ini bloklar; bu yüzden düz sözlükleri tamponlayın ve her toplu işten bir DataFrame oluşturun.

Farklı hızlarda publish yapan cihazları nasıl ele alırım?

rolling("5min", on="ts") ve resample("1min") gibi zamana dayalı pencereler kullanın. Her cihazın kaç ölçüm gönderdiğinden bağımsız olarak sabit bir süreyi kapsarlar.

Hangi zaman damgasını kullanmalıyım, payload zamanını mı yoksa varış zamanını mı?

Cihaz saati güvenilir olduğunda payload'daki cihaz zaman damgasını tercih edin, çünkü mesajlar gecikse bile doğru kalır. Payload'da zaman damgası yoksa varış zamanına başvurun.

Bu yaklaşım asyncio istemcileriyle çalışır mı?

Evet. Asenkron bir istemcide mesaj döngüsü bir coroutine olarak çalışır, ancak desen aynıdır: satırları bir tamponda toplayın ve pandas'ı sabit bir aralıkta toplu işler üzerinde çalıştırın.