Python ile MQTT Mesajlarını CSV'ye Kaydetme

MQTT mesajlarını CSV'ye kaydetmeniz gerektiğinde genellikle cevaplara hızla ulaşmak istersiniz: sensör gece boyunca ne gönderdi, bir cihaz ne sıklıkla rapor veriyor ve daha büyük bir şey kurmadan önce veri mantıklı görünüyor mu? Bu eğitimde her mesajı CSV'ye yazan, dosyaları güne ve boyuta göre döndüren ve Excel, pandas veya bir Jupyter notebook'ta sorunsuz açılan dosyalar üreten küçük bir Python abonesi geliştireceksiniz. Betiği beş dakikada ücretsiz public broker'a karşı çalıştırabilir, daha sonra kodu değiştirmeden private bir broker'a yönlendirebilirsiniz.

CSV Neden Hâlâ İyi Bir İlk Adım?

Uzun vadeli veriler için doğru hedef bir zaman serisi veritabanıdır, ancak bu her zaman doğru ilk adım değildir. CSV'nin onu erken aşama analiz için ideal kılan birkaç özelliği vardır:

Her araç okuyabilir: elektronik tablolar, pandas, R, DuckDB, hatta bir metin editörü.
Python dışında kurulacak veya işletilecek hiçbir şey yoktur.
Dosyalar kolayca kopyalanabilir, bir meslektaşa e-postayla gönderilebilir veya bir hata kaydına eklenebilir.
Yalnızca ekleme yapan yazma işlemleri basit ve sağlamdır; bir çökme en fazla son satırı kaybettirir.

Ödünleşimler de aynı derecede açıktır. CSV'de indeks, veri tipi veya eşzamanlılık kontrolü yoktur; bu yüzden birkaç yüz megabaytı aştığında yavaşlar ve hantallaşır, birden fazla sürecin yazması gerektiğinde de uygun değildir. Onu hata ayıklama, devreye alma ve kısa deneyler için bir yakalama formatı olarak görün. Kaydedici kalıcı hale geldiğinde Python ile MQTT'den SQLite'a gibi bir çözüme veya gerçek bir zaman serisi deposuna geçin.

Gerekenler

Python 3.9 veya üzeri
paho-mqtt 2.x (pip install "paho-mqtt>=2.0")
İsteğe bağlı: test mesajları göndermek için mosquitto_pub, analiz için pandas

Test için broker.connectmq.com adresindeki public MQTT test broker'ını kullanın. Boş olmayan ve benzersiz herhangi bir client ID ile anonim bağlantıları kabul eder. Public alan herkes tarafından okunabilir; bu yüzden connectmq/demo/csv-4821/ gibi rastgele bir topic öneki seçin ve oraya asla gerçek tesis veya müşteri verisi göndermeyin.

CSV Düzenini Tasarlamak

Kod yazmadan önce satır yapısına karar verin. Yaygın iki düzen vardır:

Düzen	Sütunlar	En uygun olduğu durum	Dezavantajı
Uzun (mesaj başına bir satır)	timestamp, topic, qos, retain, payload, value	Karışık topic'ler, bilinmeyen payload yapıları, hata ayıklama	Grafikler için pivot gerekir
Geniş (alan başına bir sütun)	timestamp, device, temperature, humidity...	Tek ve bilinen bir payload şeması	Yeni bir alan geldiğinde bozulur

Aşağıdaki kaydedici uzun düzeni kullanır, çünkü hiçbir zaman veri kaybetmez: hatta ne gelirse dosyaya yazılır. Ayrıca 21.4 gibi düz sayısal bir payload'dan veya value anahtarı olan bir JSON nesnesinden sayısal bir value sütunu çıkarmaya çalışır; bu da sabit bir şemaya bağlanmadan hızlı grafikler çizmeyi çok kolaylaştırır.

Zaman damgaları milisaniyeli UTC ISO 8601 formatında yazılır. Yerel saat daha okunaklı görünür, ancak yaz saati geçişlerinde yinelenen veya eksik saatler oluşturur; üstelik her analiz aracı görüntüleme sırasında UTC'yi yerel saate dönüştürebilir.

Python CSV Kaydedici

Bunu mqtt_csv_logger.py olarak kaydedin:

import csv
import json
import os
import signal
import sys
import uuid
from datetime import datetime, timezone
from pathlib import Path

import paho.mqtt.client as mqtt

BROKER = os.environ.get("MQTT_HOST", "broker.connectmq.com")
PORT = int(os.environ.get("MQTT_PORT", "1883"))
USERNAME = os.environ.get("MQTT_USERNAME")          # public broker'da None
PASSWORD = os.environ.get("MQTT_PASSWORD")
CLIENT_ID = os.environ.get("MQTT_CLIENT_ID", f"csv-logger-{uuid.uuid4().hex[:8]}")
TOPICS = os.environ.get("MQTT_TOPICS", "connectmq/demo/csv-4821/#").split(",")
OUT_DIR = Path(os.environ.get("CSV_DIR", "mqtt_logs"))
MAX_BYTES = int(os.environ.get("CSV_MAX_BYTES", str(50 * 1024 * 1024)))
SKIP_RETAINED = os.environ.get("SKIP_RETAINED", "1") == "1"

FIELDS = ["received_at", "topic", "qos", "retain", "value", "payload"]


class RotatingCsvWriter:
    """Writes rows to mqtt_YYYY-MM-DD_NNN.csv, rotating daily and by size."""

    def __init__(self, directory: Path, max_bytes: int):
        self.directory = directory
        self.max_bytes = max_bytes
        self.directory.mkdir(parents=True, exist_ok=True)
        self._file = None
        self._writer = None
        self._day = None
        self._part = 0

    def _open(self, day: str):
        if self._file:
            self._file.close()
        if day != self._day:
            self._day, self._part = day, 0
        while True:
            path = self.directory / f"mqtt_{day}_{self._part:03d}.csv"
            if not path.exists() or path.stat().st_size < self.max_bytes:
                break
            self._part += 1
        new_file = not path.exists()
        # utf-8-sig, Excel'in UTF-8'i algılaması için BOM ekler; csv için newline="" gereklidir
        self._file = open(path, "a", newline="", encoding="utf-8-sig" if new_file else "utf-8")
        self._writer = csv.DictWriter(self._file, fieldnames=FIELDS)
        if new_file:
            self._writer.writeheader()
        print(f"Writing to {path}")

    def write(self, row: dict):
        day = row["received_at"][:10]
        if self._file is None or day != self._day or self._file.tell() >= self.max_bytes:
            if self._file is not None and day == self._day:
                self._part += 1
            self._open(day)
        self._writer.writerow(row)
        self._file.flush()

    def close(self):
        if self._file:
            self._file.close()


def extract_value(text: str):
    """Return a float from '21.4' or '{"value": 21.4}', else empty string."""
    try:
        return float(text)
    except ValueError:
        pass
    try:
        data = json.loads(text)
        if isinstance(data, dict) and isinstance(data.get("value"), (int, float)):
            return data["value"]
    except json.JSONDecodeError:
        pass
    return ""


writer = RotatingCsvWriter(OUT_DIR, MAX_BYTES)


def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        print(f"Connect failed: {reason_code}")
        return
    print(f"Connected as {CLIENT_ID}")
    for topic in TOPICS:
        client.subscribe(topic.strip(), qos=1)


def on_message(client, userdata, msg):
    if SKIP_RETAINED and msg.retain:
        return
    text = msg.payload.decode("utf-8", errors="replace")
    writer.write({
        "received_at": datetime.now(timezone.utc).isoformat(timespec="milliseconds"),
        "topic": msg.topic,
        "qos": msg.qos,
        "retain": int(msg.retain),
        "value": extract_value(text),
        "payload": text,
    })


def on_disconnect(client, userdata, flags, reason_code, properties):
    print(f"Disconnected: {reason_code}, paho will reconnect")


def main():
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID)
    if USERNAME:
        client.username_pw_set(USERNAME, PASSWORD)
    if PORT == 8883:
        client.tls_set()
    client.on_connect = on_connect
    client.on_message = on_message
    client.on_disconnect = on_disconnect
    client.reconnect_delay_set(min_delay=1, max_delay=30)

    def shutdown(*_):
        client.disconnect()
        writer.close()
        sys.exit(0)

    signal.signal(signal.SIGINT, shutdown)
    signal.signal(signal.SIGTERM, shutdown)

    client.connect(BROKER, PORT, keepalive=60)
    client.loop_forever(retry_first_connection=True)


if __name__ == "__main__":
    main()

Birkaç tasarım kararını açıklamakta fayda var:

Tek iş parçacığı. loop_forever() callback'leri aynı iş parçacığında çalıştırır, dolayısıyla yazıcının kilide ihtiyacı yoktur. loop_start() kullanmaya geçerseniz write() çevresine bir threading.Lock ekleyin.
Her satırda flush. Verimden biraz yer, ancak bir elektrik kesintisinde en fazla bir satır kaybedilir. Çok yüksek mesaj hızlarında bunun yerine her N satırda bir flush yapın.
Retained message'lar varsayılan olarak atlanır. Broker her yeniden bağlantıda retained message'ları tekrar gönderir; aksi halde ağdaki her kısa kesinti yinelenen satırlar oluştururdu.
QoS 1 aboneliği. QoS 1 ile yayınlanan mesajlar en az bir kez teslim edilir, bu nedenle bir yeniden bağlantı ara sıra yinelenen bir satır üretebilir. Analiz için bu genellikle sorun değildir; bu konuyu ayrıntılı düşünmeniz gerekirse MQTT QoS seviyeleri yazısına bakın.

Çalıştırın ve Test Verisi Gönderin

Kaydediciyi başlatın:

pip install "paho-mqtt>=2.0" pandas
python mqtt_csv_logger.py

İkinci bir terminalde birkaç ölçüm yayınlayın:

for i in 1 2 3 4 5; do
  mosquitto_pub -h broker.connectmq.com -p 1883 -i pub-csv-4821 \
    -t connectmq/demo/csv-4821/line1/temp -m "{\"value\": 2$i.5, \"unit\": \"C\"}"
  mosquitto_pub -h broker.connectmq.com -p 1883 -i pub-csv-4821 \
    -t connectmq/demo/csv-4821/line1/status -m "running"
  sleep 1
done

Başlık satırı ve on satır içeren mqtt_logs/mqtt_2026-10-16_000.csv gibi bir dosya görmelisiniz. Sıcaklık satırlarında value sütununda bir sayı bulunur; durum satırlarında bu sütun boş kalır ama ham metin payload içinde korunur. Fiziksel bir cihazınız yoksa MQTT cihaz simülatörü yazısı gerçekçi veri akışlarının nasıl üretileceğini gösterir.

CSV'yi pandas ile Analiz Edin

Uzun formatlı dosyalar birkaç satırla grafiğe uygun hale getirilir:

import glob
import pandas as pd

files = sorted(glob.glob("mqtt_logs/mqtt_*.csv"))
df = pd.concat((pd.read_csv(f, encoding="utf-8-sig") for f in files), ignore_index=True)
df["received_at"] = pd.to_datetime(df["received_at"], utc=True)

numeric = df.dropna(subset=["value"])
wide = numeric.pivot_table(index="received_at", columns="topic", values="value")
print(wide.resample("1min").mean().tail())

# Topic başına mesaj hızı; çok konuşan veya sessiz cihazları fark etmek için kullanışlı
print(df.groupby("topic").size().sort_values(ascending=False))

Yalnızca mesaj hızı özeti bile çoğu zaman kaydediciyi kurmaya değer kılar: beklenenden on kat fazla yayın yapan bir cihaz veya saat 03:00'te susan bir topic hemen göze çarpar.

Sorun Giderme Kontrol Listesi

Hiçbir şey yazılmıyor. Topic filtresini kontrol edin. connectmq/demo/csv-4821/# önekin altındaki her şeyle eşleşir, ancak tek bir seviyedeki yazım hatası sessizce hiçbir şeyle eşleşmemesine yol açar. Trafiği doğrulamak için aynı filtreyle mosquitto_sub -v kullanarak abone olun.
Excel her şeyi tek sütunda gösteriyor. Bazı bölgesel ayarlar noktalı virgül bekler. Veri, Metinden/CSV'den seçeneğini kullanıp ayırıcı olarak virgülü seçin ya da bu kullanıcılar için DictWriter içinde delimiter=";" ayarlayın.
Excel'de bozuk karakterler. utf-8-sig tarafından yazılan BOM, yeni dosyalarda bu sorunu çözer. Betiğin eski sürümleriyle oluşturulan dosyaları UTF-8 seçerek yeniden içe aktarmanız gerekir.
Windows'ta satırlar arasında boş satırlar. Dosyayı newline="" olmadan açmışsınız. Yukarıdaki betik bunu zaten ele alıyor.
Kaydedicinin bağlantısı sürekli kopuyor. İki süreç aynı client ID'yi kullanıyor ve birbirini bağlantıdan atıyor. Betiğin rastgele bir ID üretmesine izin verin veya benzersiz bir MQTT_CLIENT_ID belirleyin.
İkili payload'lar anlamsız görünüyor. Protobuf veya sıkıştırılmış payload'lar metin değildir. Bu topic'ler için çözümlenmiş metin yerine msg.payload.hex() değerini saklayın.

Private Bir Broker'a Geçiş

Public broker kaydediciyi denemek için mükemmeldir, ancak gerçek veriler kimlik bilgileri olan private bir broker'da olmalıdır. ConnectMQ ile portalda bir broker oluşturur ve otomatik üretilmiş bir kullanıcı adı, parola ve client ID alırsınız. Ardından aynı betiği ortam değişkenleri ve 8883 portunda TLS ile çalıştırın:

export MQTT_HOST=YOUR_BROKER_HOST
export MQTT_PORT=8883
export MQTT_USERNAME=YOUR_USERNAME
export MQTT_PASSWORD=YOUR_PASSWORD
export MQTT_CLIENT_ID=YOUR_CLIENT_ID
export MQTT_TOPICS="factory1/#"
python mqtt_csv_logger.py

ConnectMQ'da client ID'ler verilen kimlik bilgilerine bağlı olduğundan rastgele bir ID yerine portaldaki client ID'yi kullanın. Yönetilen bir MQTT broker aynı zamanda kaydedicinin sıradan bir client olması anlamına gelir: yanında Mosquitto çalıştırmanız gerekmez ve portalın canlı topic gezgini, dosyaya güvenmeden önce kaydedicinin neler alması gerektiğini kontrol etmenizi sağlar.

CSV Ne Zaman Yetersiz Kalır?

CSV kaydı bir basamak taşıdır. Şunlardan herhangi biri gerçekleştiğinde bir sonraki adıma geçin: dosyalar Excel'in rahatça açabileceği boyutu aşıyorsa, birden fazla sürecin yazması gerekiyorsa, haftalarca veriyi kapsayan sorgulara ihtiyacınız varsa veya başkalarının canlı panolara ihtiyacı varsa. Bu serideki doğal sonraki adımlar, paylaşılan hafif raporlama için MQTT verisini Google Sheets'e göndermek ve uzun ömürlü her şey için gerçek bir veritabanıdır. Bu entegrasyonlarda kullanabileceğiniz Python, JavaScript ve diğer client'lara genel bir bakış için MQTT client kütüphaneleri rehberine göz atın. paho-mqtt dokümantasyonu burada kullanılan tüm client seçeneklerini kapsar.

ConnectMQ ile Başlayın

CSV kaydedicinizi birkaç dakika içinde TLS ile korunan private bir broker'a yönlendirin. app.connectmq.com adresinden kaydolun, kredi kartı gerektirmeyen ücretsiz Serverless planını seçin, bir broker oluşturun ve üretilen kimlik bilgilerini yukarıdaki ortam değişkenlerine yapıştırın. Ücretsiz planda 25 bağlantıya kadar hakkınız olur; bu, bir kaydedici ve bir avuç cihaz için fazlasıyla yeterlidir. Ayrıca test mesajları göndermek için tarayıcı içi bir MQTT client'ı da sunulur.

Sık Sorulan Sorular

MQTT mesajlarını kod yazmadan CSV'ye kaydedebilir miyim?

Evet, Node-RED veya Telegraf gibi araçlar dosya yazabilir, ancak kısa bir Python betiği sütunlar, dosya döndürme ve retained message yönetimi üzerinde tam kontrol sağlar. Bu yazıdaki betik 150 satırın altındadır ve tek bir bağımlılığı vardır.

Kaydedici çevrimdışıyken mesaj kaybeder miyim?

Clean session ile evet: kaydedici bağlı değilken yayınlanan mesajlar onun için saklanmaz. QoS 1 ile kalıcı bir oturum kullanmak, broker'ın mesajları bir süre kuyrukta tutmasını sağlar; ancak garantili bir geçmiş için uzun süre çalışan bir servisin beslediği bir veritabanı kullanın.

CSV dosyaları ne kadar büyüyebilir?

Betik varsayılan olarak 50 MB'ta dosyayı döndürür ve her UTC gününde yeni bir dosya başlatır. CSV_MAX_BYTES değerini analiz için kullandığınız araca göre ayarlayın; elektronik tablolar pandas'tan çok daha önce ağırlaşır.

Public broker'dan veri kaydetmek güvenli mi?

Test verileri için güvenlidir. broker.connectmq.com'un public alanındaki her şey, aynı topic'e abone olan herkes tarafından okunabilir; bu yüzden gerçek ölçümler için kimlik bilgileri olan private bir broker kullanın.

Hızlı Analiz İçin Python ile MQTT Mesajlarını CSV'ye Kaydetme