MQTT Verisini TimescaleDB'ye Kaydetme: IoT Zaman Serileri

MQTT verisini TimescaleDB'ye göndermek, size PostgreSQL'in güvenilirliğini ve SQL'ini zaman serileri için tasarlanmış depolama ve sorgularla birlikte sunar. Bu eğitimde bir bulut broker'dan gelen cihaz mesajlarını toplu halde bir TimescaleDB hypertable'ına yazan bir Python subscriber geliştirecek, ardından veri büyüdükçe panoların hızlı kalması için continuous aggregate'ler, sıkıştırma ve saklama politikaları ekleyeceksiniz.

Mimari

devices --MQTT--> broker (ConnectMQ) --MQTT--> Python bridge --COPY--> TimescaleDB --SQL--> Grafana / apps

Broker mesajları yönlendirir; veritabanlarına yazmaz. Köprü (bridge), subscribe olan, ayrıştıran ve ekleyen sıradan bir MQTT client'tır. Bu ayrım bir güçtür: cihazlara dokunmadan köprüyü yeniden başlatabilir, ölçekleyebilir veya değiştirebilirsiniz; aynı desen herhangi bir yönetilen MQTT broker ile çalışır. MQTT tarafını ConnectMQ üstlenir; TimescaleDB ve köprü ise verinizi nerede barındırıyorsanız orada çalışır.

Bileşen	Rol	Bu eğitimde
Cihazlar	JSON telemetri publish eder	`mosquitto_pub` ile simüle edilir
Broker	Mesajları yönlendirir, oturumları tutar	Önce broker.connectmq.com, ardından özel bir broker
Köprü	Subscribe olur, doğrular, toplar, ekler	Python, paho-mqtt 2.x, psycopg 3
Veritabanı	Zaman serilerini saklar ve özetler	PostgreSQL 16 üzerinde TimescaleDB

Adım 1: TimescaleDB'yi Çalıştırın

En hızlı başlangıç resmi Docker imajıdır:

docker run -d --name tsdb -p 5432:5432 \
  -e POSTGRES_PASSWORD=change-me \
  -v tsdb-data:/var/lib/postgresql/data \
  timescale/timescaledb:latest-pg16

psql -h localhost -U postgres ile bağlanın ve şemayı oluşturun.

Adım 2: Hypertable'ı Tasarlayın

IoT cihazları nadiren aynı alan setini publish eder. Metrik değeri başına bir satır içeren dar bir tablo, şema değişikliği gerektirmeden yeni sensörleri karşılar ve çok iyi sıkışır.

CREATE EXTENSION IF NOT EXISTS timescaledb;

CREATE TABLE readings (
    time       TIMESTAMPTZ      NOT NULL,
    device_id  TEXT             NOT NULL,
    metric     TEXT             NOT NULL,
    value      DOUBLE PRECISION NOT NULL
);

SELECT create_hypertable('readings', by_range('time', INTERVAL '1 day'));

CREATE INDEX readings_device_metric_time
    ON readings (device_id, metric, time DESC);

create_hypertable, tabloyu zamana göre bölümlenmiş bir dizi chunk'a dönüştürür; burada her gün için bir chunk. Zamana göre filtreleyen sorgular yalnızca ilgili chunk'lara dokunur; eski chunk'lar da tek bir ucuz işlemle sıkıştırılabilir veya silinebilir. Chunk aralığını en yeni chunk'ların belleğe rahatça sığacağı şekilde seçin; bir gün çoğu küçük ve orta ölçekli filoya uyar.

Her cihaz tam olarak aynı alanları gönderiyorsa geniş bir tablo (time, device_id, temperature, humidity, ...) da uygundur ve sorgulaması biraz daha basittir. Aşağıdaki köprü her iki duruma da kolayca uyarlanabilir.

Adım 3: MQTT'den TimescaleDB'ye Köprü

Bağımlılıkları kurun:

pip install "paho-mqtt>=2,<3" "psycopg[binary]>=3.1"

Cihazlar <prefix>/<device_id>/telemetry topic'ine şuna benzer JSON publish eder:

{"ts": 1792137600.5, "temperature": 23.4, "humidity": 41.0, "battery": 3.71}

Köprünün iki thread'i vardır. MQTT ağ thread'i yalnızca ayrıştırır ve satırları kuyruğa ekler; bu sayede asla veritabanında bloklanmaz. Bir yazıcı thread'i kuyruğu boşaltır ve hangisi önce gelirse, her saniyede veya her 1.000 satırda bir PostgreSQL COPY ile bir toplu yazma yapar.

mqtt_to_timescaledb.py:

import json
import logging
import os
import queue
import signal
import threading
import time
import uuid
from datetime import datetime, timezone

import paho.mqtt.client as mqtt
import psycopg
from paho.mqtt.packettypes import PacketTypes
from paho.mqtt.properties import Properties

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("bridge")

MQTT_HOST = os.getenv("MQTT_HOST", "broker.connectmq.com")
MQTT_PORT = int(os.getenv("MQTT_PORT", "8883"))
MQTT_USER = os.getenv("MQTT_USERNAME")
MQTT_PASS = os.getenv("MQTT_PASSWORD")
CLIENT_ID = os.getenv("MQTT_CLIENT_ID") or f"tsdb-bridge-{uuid.uuid4().hex[:8]}"
PREFIX = os.getenv("MQTT_TOPIC_PREFIX", "connectmq/demo/tsdb-5b7e21")
DSN = os.getenv("PG_DSN", "postgresql://postgres:change-me@localhost:5432/postgres")

BATCH_SIZE = 1000
FLUSH_SECONDS = 1.0
MAX_QUEUE = 100_000

rows: "queue.Queue[tuple]" = queue.Queue(maxsize=MAX_QUEUE)
stop = threading.Event()


def parse(topic: str, payload: bytes):
    """Tek bir MQTT mesajını (time, device_id, metric, value) satırlarına dönüştür."""
    parts = topic.split("/")
    if len(parts) < 2 or parts[-1] != "telemetry":
        return []
    device_id = parts[-2]
    data = json.loads(payload)
    if not isinstance(data, dict):
        return []
    ts = data.pop("ts", None)
    when = (datetime.fromtimestamp(float(ts), tz=timezone.utc)
            if isinstance(ts, (int, float)) else datetime.now(timezone.utc))
    out = []
    for metric, value in data.items():
        if isinstance(value, bool):
            value = float(value)
        if isinstance(value, (int, float)):
            out.append((when, device_id, metric[:64], float(value)))
    return out


def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        log.error("connect failed: %s", reason_code)
        return
    log.info("connected (session present: %s)", flags.session_present)
    client.subscribe(f"{PREFIX}/+/telemetry", qos=1)


def on_message(client, userdata, msg):
    try:
        for row in parse(msg.topic, msg.payload):
            rows.put_nowait(row)
    except (ValueError, json.JSONDecodeError):
        log.warning("bad payload on %s", msg.topic)
    except queue.Full:
        log.error("queue full, dropping message from %s", msg.topic)


def writer():
    conn = None
    batch = []
    last_flush = time.monotonic()
    while not (stop.is_set() and rows.empty() and not batch):
        try:
            batch.append(rows.get(timeout=0.2))
            if len(batch) < BATCH_SIZE and time.monotonic() - last_flush < FLUSH_SECONDS:
                continue
        except queue.Empty:
            if not batch:
                continue
        try:
            if conn is None or conn.closed:
                conn = psycopg.connect(DSN, autocommit=False)
            with conn.cursor() as cur:
                with cur.copy("COPY readings (time, device_id, metric, value) FROM STDIN") as cp:
                    for r in batch:
                        cp.write_row(r)
            conn.commit()
            log.info("wrote %d rows", len(batch))
            batch.clear()
            last_flush = time.monotonic()
        except psycopg.Error as e:
            log.error("database error, retrying in 5 s: %s", e)
            if conn is not None:
                conn.close()
            conn = None
            time.sleep(5)  # toplu veriyi koru ve yeniden dene
    if conn is not None:
        conn.close()


def main():
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID,
                         protocol=mqtt.MQTTv5)
    if MQTT_USER:
        client.username_pw_set(MQTT_USER, MQTT_PASS)
    if MQTT_PORT == 8883:
        client.tls_set()
    client.on_connect = on_connect
    client.on_message = on_message
    client.reconnect_delay_set(min_delay=1, max_delay=60)

    # Kalıcı oturum: köprü yeniden başlarken broker QoS 1 mesajlarını kuyrukta tutar
    props = Properties(PacketTypes.CONNECT)
    props.SessionExpiryInterval = 1800

    t = threading.Thread(target=writer, daemon=True)
    t.start()

    def shutdown(*_):
        log.info("shutting down")
        client.disconnect()
        stop.set()

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

    client.connect(MQTT_HOST, MQTT_PORT, keepalive=30, clean_start=False, properties=props)
    client.loop_forever(retry_first_connection=True)
    t.join(timeout=30)


if __name__ == "__main__":
    main()

Neden Bu Şekilde Tasarlandı?

Satır satır INSERT yerine COPY. Toplu COPY, PostgreSQL'e çok sayıda satır yüklemenin en hızlı yoludur ve transaction sayısını düşük tutar.
Sınırlı bir kuyruk. Veritabanı uzun süre kapalı kalırsa bellek sınırsızca büyümez. Köprü çökmek yerine günlüğe yazar ve mesajı bırakır; MAX_QUEUE değerini atlatmak istediğiniz kesinti süresine göre boyutlandırın.
Kalıcı MQTT oturumu. clean_start=False, sabit bir client ID ve bir oturum süresi (session expiry) ile broker, köprü yeniden başlarken kuyruktaki QoS 1 mesajlarını tutar. Public broker'da oturumlar en fazla 30 dakika sürer. Bunun çalışması için sabit bir MQTT_CLIENT_ID ayarlayın; rastgele bir ID her seferinde yeni bir oturum oluşturur. Arka plan bilgisi MQTT kalıcı oturumları yazısındadır.
Mümkün olduğunda cihaz zaman damgaları. Çevrimdışıyken okumaları tamponlayan cihazlar bunları geç gönderir; kendi ts değerlerini kullanmak zaman serisini doğru tutar.

Adım 4: Pipeline'ı Test Edin

Köprüyü başlatın, ardından public MQTT test broker'ına birkaç okuma publish edin. Public alan herkes tarafından okunabilir; bu yüzden yapay veri kullanın:

python mqtt_to_timescaledb.py

# başka bir terminalde
for i in $(seq 1 20); do
  mosquitto_pub -h broker.connectmq.com -p 1883 -i tsdb-sim-01 -q 1 \
    -t "connectmq/demo/tsdb-5b7e21/sensor-01/telemetry" \
    -m "{\"temperature\": 2$((RANDOM % 10)).$((RANDOM % 10)), \"humidity\": 4$((RANDOM % 10))}"
  sleep 1
done

Veriyi kontrol edin:

SELECT device_id, metric, count(*), min(time), max(time)
FROM readings GROUP BY device_id, metric;

Adım 5: Hızlı Panolar için Continuous Aggregate'ler

Bir aylık ham, saniye başına veriyi çizen panolar yavaştır. Bir continuous aggregate, özetleri önceden hesaplar ve arka planda artımlı olarak yeniler:

CREATE MATERIALIZED VIEW readings_1m
WITH (timescaledb.continuous) AS
SELECT time_bucket(INTERVAL '1 minute', time) AS bucket,
       device_id,
       metric,
       avg(value) AS avg_value,
       min(value) AS min_value,
       max(value) AS max_value,
       count(*)   AS samples
FROM readings
GROUP BY bucket, device_id, metric
WITH NO DATA;

SELECT add_continuous_aggregate_policy('readings_1m',
    start_offset      => INTERVAL '2 hours',
    end_offset        => INTERVAL '1 minute',
    schedule_interval => INTERVAL '1 minute');

start_offset, her yenilemede geç gelen verinin ne kadar geriye kadar alınacağını belirler. Cihazlar iki saatten uzun süre çevrimdışı kalıp ardından tamponlanmış okumaları yükleyebiliyorsa bu değeri genişletin. View'u tam olarak bir tablo gibi sorgulayın:

SELECT bucket, avg_value
FROM readings_1m
WHERE device_id = 'sensor-01' AND metric = 'temperature'
  AND bucket > now() - INTERVAL '24 hours'
ORDER BY bucket;

Adım 6: Sıkıştırma ve Saklama

Sensör verisi, cihaza ve metriğe göre gruplandığında son derece iyi sıkışır:

ALTER TABLE readings SET (
    timescaledb.compress,
    timescaledb.compress_segmentby = 'device_id, metric',
    timescaledb.compress_orderby   = 'time DESC'
);

SELECT add_compression_policy('readings', INTERVAL '7 days');
SELECT add_retention_policy('readings', INTERVAL '90 days');

Bu yapılandırma 90 günlük ham veriyi saklar, bir haftadan eski her şeyi sıkıştırır ve bir dakikalık özetin uzun vadeli eğilimler için yaşamaya devam etmesini sağlar. Yeni TimescaleDB sürümleri sıkıştırmayı "columnstore" olarak da adlandırır; yukarıdaki politikalar çalışmaya devam eder.

Veriyi Grafana'da Görselleştirme

Grafana'nın yerleşik PostgreSQL veri kaynağı TimescaleDB ile değiştirilmeden çalışır. Veritabanını gösteren bir veri kaynağı ekleyin, ayarlarında TimescaleDB seçeneğini etkinleştirin ve köprünün kullandığı superuser yerine salt okunur bir rolle bağlanın. Panel sorgularını continuous aggregate'e karşı yazın ve Grafana makrolarının panonun zaman aralığını eklemesine izin verin:

SELECT bucket AS "time", device_id AS metric, avg_value
FROM readings_1m
WHERE $__timeFilter(bucket) AND metric = 'temperature'
ORDER BY bucket;

Ham readings tablosunu yalnızca saniye düzeyindeki ayrıntının önemli olduğu kısa aralıklar için kullanın; günlere yayılan her şey aggregate'e gitmelidir. Hızlı bir canlı görünüm için veritabanını atlamayı tercih ederseniz, MQTT veri kaynağıyla Grafana topic'leri doğrudan çizebilir, ancak geçmiş tutmaz.

Üretim Kontrol Listesi

Public broker'dan özel bir broker'a geçin: ConnectMQ portalında bir broker oluşturun, ardından MQTT_USERNAME, MQTT_PASSWORD ve MQTT_CLIENT_ID değerlerini üretilen değerlere ayarlayın.
8883 portunda TLS'i koruyun.
Köprüyü systemd altında veya her zaman yeniden başlayan (restart always) bir container olarak çalıştırın.
Uçtan uca QoS 1 kullanın; ücretsiz plan QoS 1'e kadar destekler.
Yinelemeler önemliyse yazmaları idempotent yapın: QoS 1 yeniden teslim edebilir. (device_id, metric, time) üzerinde benzersiz bir indeks ile ON CONFLICT DO NOTHING kullanan bir hazırlık (staging) tablosu bunu çözer.
Köprü günlüklerinde kuyruk derinliğini ve saniyedeki satır sayısını izleyin.
Yatay ölçeklemek için ücretli ConnectMQ Serverless planları ve üzerinde bulunan MQTT 5 shared subscription'larıyla ($share/tsdb/<prefix>/+/telemetry) birkaç köprü çalıştırın. Bkz. MQTT shared subscription'ları.

Sorun Giderme

Belirti	Olası neden
`relation "readings" does not exist`	Şema, `PG_DSN`'in gösterdiğinden farklı bir veritabanında oluşturulmuş
Köprü bağlanıyor ama hiçbir şey yazmıyor	Topic `/telemetry` ile bitmiyor veya önek uyuşmuyor
Satırların zamanı yanlış	Cihaz saati yanlış; çok ileri tarihli `ts` değerlerini yok saymayı düşünün
Yeniden başlatmalardan sonra boşluklar	Rastgele client ID, dolayısıyla kalıcı oturum yok
Aggregate boş	`WITH NO DATA` kullanıldı ve politika henüz çalışmadı; `CALL refresh_continuous_aggregate('readings_1m', NULL, NULL);` çağırın

Daha basit bir depolama ile başlamayı tercih ederseniz, bu yaklaşımı MQTT mesajlarını MongoDB'ye kaydetme veya MQTT mesajlarını CSV'ye kaydetme ile karşılaştırın. Daha fazla client deseni için MQTT client kütüphaneleri rehberi iyi bir sonraki okumadır.

ConnectMQ ile Başlayın

TimescaleDB köprünüzü özel bir broker'a yönlendirin. app.connectmq.com üzerinde kredi kartı gerekmeden ücretsiz bir Serverless broker oluşturun, üretilen kimlik bilgilerini ortam değişkenleri olarak ayarlayın ve cihazlarınızın publish ettiği verileri veritabanına ulaşmadan önce portalın canlı topic explorer'ı ile doğrulayın.

Sık Sorulan Sorular

ConnectMQ, MQTT verisini doğrudan TimescaleDB'ye yazabilir mi?

Hayır, ConnectMQ bir MQTT broker'dır ve veritabanı entegrasyonları içermez. Veriyi yazmak için bu eğitimdeki Python köprüsü, Telegraf veya Node-RED gibi bir subscriber çalıştırın.

TimescaleDB'de MQTT verisi için dar mı yoksa geniş tablo mu kullanmalıyım?

Cihazlar değişken alanlar gönderiyorsa dar bir tablo (time, device, metric, value), her cihaz aynı sabit seti gönderiyorsa geniş bir tablo kullanın. Dar tablolar, sensör eklediğinizde migration gerektirmez.

Veritabanı kapalıyken MQTT mesajlarını kaybetmekten nasıl kaçınırım?

Köprüde sınırlı bir kuyrukla tamponlayın ve yazmaları yeniden deneyin; ayrıca broker'ın köprü yeniden başlarken mesajları tutması için QoS 1 ile kalıcı bir MQTT oturumu kullanın.

TimescaleDB yerine düz PostgreSQL yeterli mi?

Küçük hacimler için evet. TimescaleDB, veri büyüdükçe değer kazanan otomatik bölümleme, continuous aggregate'ler, sıkıştırma ve saklama politikaları ekler.

MQTT Verisini TimescaleDB'ye Kaydetme: PostgreSQL'de IoT Zaman Serileri