Python ile MQTT verisini SQLite'a göndermek, IoT projenize bir geçmiş kazandırmanın en basit yoludur: tek bir dosya, veritabanı sunucusu yok ve sorgular için tam SQL. Bu eğitimde broker.connectmq.com üzerinden gelen her mesajı zaman damgası, topic ve bayraklarıyla saklayan, verimli toplu yazmalar yapan ve yeniden bağlanmalara dayanan bir paho-mqtt subscriber oluşturacaksınız. Ardından verileri son değerler, saatlik ortalamalar ve saklama süresi temizliği için sorgulayacaksınız.
MQTT Verisi için Neden SQLite?
Bir MQTT broker, mesajları o anda subscribe olan kişiye iletir. Bir veritabanı değildir ve topic başına bir retained mesaj dışında geçmiş tutmaz. "Gece saat 3'te sıcaklık neydi?" sorusunu yanıtlamak istiyorsanız, bir şeyin subscribe olup veriyi saklaması gerekir.
SQLite, bu iş için mükemmel bir ilk tercihtir:
| İhtiyaç | SQLite bunu nasıl karşılar |
|---|---|
| Sıfır kurulum | Python'un standart kütüphanesinde yerleşik (sqlite3) |
| Tek dosya | Yedeklemesi, kopyalaması veya bir GUI tarayıcıda açması kolay |
| SQL sorguları | Filtreleme, gruplama, JSON fonksiyonları, tarih fonksiyonları |
| Orta düzey yazma hızları | Transaction'lar içinde toplu yapıldığında saniyede binlerce insert |
| Edge cihazlar | Bir Raspberry Pi veya endüstriyel PC'de rahatça çalışır |
Farklı makinelerdeki çok sayıda eş zamanlı yazıcı veya bir cihaz filosundan gelen yıllarca yüksek frekanslı veri için doğru araç değildir. Bu durumlar için MQTT'den TimescaleDB'ye rehberine veya MQTT verisini InfluxDB'de saklama yazısına bakın. Tek bir saha, bir gateway veya bir prototip için SQLite'ı geçmek zordur.
Mimari: Ağ Thread'ini Serbest Tutun
Saf yaklaşım, on_message içinde bir INSERT ve bir commit() çalıştırır. Bu dakikada birkaç mesajda işe yarar, ancak her commit bir disk senkronizasyonunu zorlar. Daha yüksek hızlarda callback geride kalır, paho'nun network loop'u takılır ve keepalive ping'leri broker'ın bağlantıyı düşürmesine yetecek kadar gecikebilir.
Sağlam tasarım iki işi birbirinden ayırır:
- paho, network loop'unu bir arka plan thread'inde çalıştırır (
loop_start()). on_messagemesajı birqueue.Queueiçine koymaktan başka bir şey yapmaz.- SQLite bağlantısının sahibi olan ana thread mesajları kuyruktan çeker ve toplu olarak yazar; her toplu işte veya saniyede bir kez commit eder.
Bu aynı zamanda SQLite'ın, bir bağlantının onu oluşturan thread'den kullanılması gerektiği kuralını da aşmanızı sağlar.
Adım 1: Kurulum ve Hazırlık
Yalnızca paho-mqtt gerekir; sqlite3 Python ile birlikte gelir:
python -m venv .venv
source .venv/bin/activate # Windows: .venv\Scripts\activate
pip install "paho-mqtt>=2.0"
Verilerinizin herkese açık MQTT test broker'ının diğer kullanıcılarının verileriyle karışmaması için rastgele bir topic öneki seçin. Bu eğitim connectmq/demo/k7f3q2 kullanır.
Adım 2: Python ile MQTT'den SQLite'a Logger
Bunu mqtt_sqlite_logger.py olarak kaydedin:
import queue
import signal
import sqlite3
import time
import uuid
from datetime import datetime, timezone
import paho.mqtt.client as mqtt
BROKER = "broker.connectmq.com"
PORT = 1883
TOPIC = "connectmq/demo/k7f3q2/#"
DB_PATH = "mqtt_messages.db"
BATCH_SIZE = 200
FLUSH_INTERVAL = 1.0 # saniye
SCHEMA = """
CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY,
received_at TEXT NOT NULL,
topic TEXT NOT NULL,
payload TEXT,
qos INTEGER NOT NULL,
retained INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_messages_topic_time ON messages (topic, received_at);
CREATE INDEX IF NOT EXISTS idx_messages_time ON messages (received_at);
"""
inbox: "queue.Queue[tuple]" = queue.Queue(maxsize=50_000)
running = True
def utc_now() -> str:
# 2026-10-16T08:15:30.123Z, sıralanabilir ve SQLite tarih fonksiyonlarınca anlaşılır
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
def on_connect(client, userdata, flags, reason_code, properties):
if reason_code.is_failure:
print(f"Connect failed: {reason_code}")
return
print(f"Connected, subscribing to {TOPIC}")
client.subscribe(TOPIC, qos=1)
def on_message(client, userdata, msg):
row = (
utc_now(),
msg.topic,
msg.payload.decode("utf-8", errors="replace"),
msg.qos,
int(msg.retain),
)
try:
inbox.put_nowait(row)
except queue.Full:
print("Queue full, dropping message") # veritabanı yetişemiyor
def on_disconnect(client, userdata, disconnect_flags, reason_code, properties):
print(f"Disconnected: {reason_code}, paho will reconnect")
def open_db(path: str) -> sqlite3.Connection:
db = sqlite3.connect(path)
db.execute("PRAGMA journal_mode=WAL") # okuyucular yazıcıyı bloklamaz
db.execute("PRAGMA synchronous=NORMAL") # WAL ile güvenli, çok daha hızlı
db.executescript(SCHEMA)
return db
def flush(db: sqlite3.Connection, batch: list) -> None:
if not batch:
return
with db: # toplu iş başına bir transaction
db.executemany(
"INSERT INTO messages (received_at, topic, payload, qos, retained) VALUES (?, ?, ?, ?, ?)",
batch,
)
print(f"Stored {len(batch)} messages")
batch.clear()
def stop(*_):
global running
running = False
def main():
signal.signal(signal.SIGINT, stop)
signal.signal(signal.SIGTERM, stop)
db = open_db(DB_PATH)
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"sqlite-{uuid.uuid4().hex[:8]}")
# Özel ConnectMQ broker:
# client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD")
# client.tls_set(); ve 8883 portunu kullanın
client.on_connect = on_connect
client.on_message = on_message
client.on_disconnect = on_disconnect
client.reconnect_delay_set(min_delay=1, max_delay=60)
client.connect_async(BROKER, PORT, keepalive=60)
client.loop_start()
batch: list = []
last_flush = time.monotonic()
try:
while running:
try:
batch.append(inbox.get(timeout=0.2))
except queue.Empty:
pass
if len(batch) >= BATCH_SIZE or time.monotonic() - last_flush >= FLUSH_INTERVAL:
flush(db, batch)
last_flush = time.monotonic()
finally:
client.disconnect()
client.loop_stop()
while not inbox.empty():
batch.append(inbox.get_nowait())
flush(db, batch)
db.close()
print("Logger stopped cleanly")
if __name__ == "__main__":
main()
python mqtt_sqlite_logger.py ile çalıştırın. Başka bir terminalde biraz test verisi publish edin:
for i in $(seq 1 20); do
mosquitto_pub -h broker.connectmq.com -p 1883 -i sqlite-test-pub \
-t "connectmq/demo/k7f3q2/sensors/room$((i % 2 + 1))" \
-m "{\"temperature\": $((20 + RANDOM % 5)).$((RANDOM % 10)), \"humidity\": $((40 + RANDOM % 10))}"
sleep 1
done
Logger yaklaşık saniyede bir "Stored 1 messages" yazdırır. Ctrl+C ile durdurun; kuyrukta kalanlar çıkıştan önce yazılır.
Logger Nasıl Güvenilir Kalıyor
Zaman damgaları. Logger, mesajın alındığı anı milisaniye hassasiyetinde UTC olarak kaydeder. Cihaz saatleri kayar ve bazı cihazların saati hiç yoktur; bu yüzden alınma zaman damgası güvenilir bir temel çizgidir. Payload'ınız kendi zaman damgasını taşıyorsa ikisini birden saklayın. Python MQTT JSON rehberi publish tarafında nasıl zaman damgası ekleneceğini gösterir.
WAL modu. Write-ahead logging, logger yazarken bir panelin veya sqlite3 CLI'ın veritabanını okumasına izin verir. Bu olmadan uzun okumalar insert'leri bloklayabilir.
Sınırlı kuyruk. Disk takılırsa kuyruk 50.000 mesaja kadar büyür, sonrasında belleği tüketmek yerine yeni mesajları bir uyarıyla düşürür. Boyutu trafiğinize göre ayarlayın.
Yeniden bağlanmalar. connect_async() ile loop_start() arka planda yeniden denemeyi sürdürür ve abonelik on_connect içinde geri yüklenir. Logger çevrimdışıyken publish edilen mesajlar, kalıcı bir oturum kullanmadığınız sürece kaybolur. MQTT 5 ile bir session expiry interval ve clean_start=False ayarlayarak broker'ın QoS 1 mesajlarını sizin için kuyruğa almasını sağlayabilirsiniz; herkese açık broker oturumları 30 dakikaya kadar tutar. paho-mqtt yeniden bağlanma rehberi ayrıntıları açıklar.
Retained mesajlar. Her yeniden bağlanmada broker retained mesajları tekrar gönderir; bu yüzden onların birden fazla kez saklandığını görürsünüz. retained sütunu bunları analizden filtrelemenizi sağlar.
Adım 3: MQTT Geçmişinizi Sorgulayın
Veritabanını SQLite CLI (sqlite3 mqtt_messages.db) veya herhangi bir SQLite tarayıcısıyla açın ve şu sorguları deneyin.
Bir topic için son 10 mesaj:
SELECT received_at, payload
FROM messages
WHERE topic = 'connectmq/demo/k7f3q2/sensors/room1'
ORDER BY received_at DESC
LIMIT 10;
SQLite'ın MAX() ile çıplak sütunlar için belgelenmiş davranışını kullanarak topic başına en son değer:
SELECT topic, payload, MAX(received_at) AS last_seen
FROM messages
WHERE retained = 0
GROUP BY topic;
SQLite'ın yerleşik JSON fonksiyonlarını kullanarak JSON payload'lardan saatlik ortalamalar:
SELECT strftime('%Y-%m-%d %H:00', received_at) AS hour,
topic,
ROUND(AVG(json_extract(payload, '$.temperature')), 2) AS avg_temperature,
COUNT(*) AS samples
FROM messages
WHERE topic LIKE 'connectmq/demo/k7f3q2/sensors/%'
AND json_valid(payload)
GROUP BY hour, topic
ORDER BY hour;
Yalnızca son yedi günü tutun ve alanı geri kazanın:
DELETE FROM messages
WHERE received_at < strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-7 days');
VACUUM;
received_at sabit bir ISO 8601 biçiminde saklandığından, düz dizgi karşılaştırması doğru sıralar ve zaman aralığı filtresi index'i kullanabilir. Saklama sorgusunu cron'dan veya günde bir kez logger'ın kendisinden çalıştırabilirsiniz.
Verilerle Python'da çalışmak için pandas.read_sql_query("SELECT ...", sqlite3.connect("mqtt_messages.db")) bir DataFrame döndürür. MQTT ve pandas eğitimi analiz ve yeniden örneklemeyi ele alır.
Sorun Giderme Kontrol Listesi
- "database is locked": başka bir işlem bir yazma transaction'ı tutuyor. WAL modunun etkin olduğundan ve dosyaya yalnızca bir logger'ın yazdığından emin olun.
- Hiçbir şey saklanmıyor: topic filtresinin publisher'ın topic'iyle eşleştiğini doğrulayın;
mosquitto_sub -h broker.connectmq.com -p 1883 -i check-k7f3q2 -t 'connectmq/demo/k7f3q2/#' -vile test edin. json_extractNULL döndürüyor: anahtar adı veya JSON yolu farklı ya da payload JSON değil. Önce birkaç ham satırı inceleyin.- Veritabanı hızla büyüyor: saklama işini ekleyin ve publisher tarafında yalnızca değişen değerleri saklamayı düşünün.
- Logger mesaj düşürüyor:
BATCH_SIZEdeğerini artırın, veritabanını daha hızlı bir depolamaya taşıyın veya aboneliği gerçekten ihtiyaç duyduğunuz topic'lerle sınırlayın.
Prototipten Üretime
Gerçek kurulumlarda logger'ı yeniden başlatmalardan sonra tekrar ayağa kalkacak şekilde bir servis olarak (Linux'ta systemd) çalıştırın ve herkese açık broker yerine özel bir broker'a yönlendirin. ConnectMQ ile portalda bir broker oluşturur, üretilen kullanıcı adı, parola ve client ID'yi kopyalar ve 8883 portunda TLS'i etkinleştirirsiniz; Python MQTT TLS eğitimi tam ayarları gösterir. ConnectMQ broker'ı mesaj geçmişini kendisi saklamaz; bu nedenle bunun gibi harici bir subscriber dürüst ve önerilen desendir. İstemci kodu için bir başlangıç noktasına ihtiyacınız varsa paho mqtt python eğitimine, diğer diller için MQTT istemci kütüphanelerine genel bakış yazısına bakın. Python sqlite3 dokümantasyonu transaction'ları ve bağlantı seçeneklerini derinlemesine ele alır.
ConnectMQ ile Başlayın
ConnectMQ; MQTT 3.1.1 ve 5.0, TLS ve WebSocket sunan, ayrıca cihazlarınızın gönderdiklerini kaydetmeden önce kontrol edebilmeniz için canlı bir topic gezgini içeren barındırılan bir MQTT broker hizmetidir. Bugün broker.connectmq.com üzerinden kayıt almaya başlayın, ardından kendi bulut MQTT broker'ınıza geçin. Serverless planında ücretsiz hesap oluşturun, kredi kartı gerekmez.
Sık Sorulan Sorular
SQLite sürekli MQTT verisini kaldırabilir mi?
Evet, orta düzey hızlarda tek bir logger için. Insert'leri transaction'lar içinde toplu yapmak ve WAL modunu etkinleştirmek, SQLite'ın mütevazı donanımda birçok sensörden gelen sürekli bir akışı karşılamasını sağlar.
Neden doğrudan on_message içinde insert yapmıyoruz?
Callback içinde commit yapmak, paho'nun network loop'unu disk G/Ç'si üzerinde bloklar. Yük altında bu, keepalive ping'lerini ve onayları geciktirir; bu yüzden bir kuyruk ve ayrı bir yazıcı daha güvenilirdir.
Logger yeniden başlarken mesaj kaybetmekten nasıl kaçınırım?
Kalıcı bir MQTT 5 oturumuyla (clean_start=False ve bir session expiry interval) QoS 1 kullanın; böylece broker, oturum süresi sınırı dahilinde siz çevrimdışıyken mesajları kuyruğa alır.
SQLite'ta binary payload saklayabilir miyim?
Evet. payload sütununu BLOB olarak değiştirin ve metne çözmek yerine msg.payload değerini doğrudan ekleyin.
