Node-RED MySQL: Parametreli Sorgularla MQTT Verilerini MySQL veya MariaDB'ye Kaydetme

Node-RED ve MySQL, MQTT sensör verilerini ekibinizin zaten sorgulamayı, yedeklemeyi ve raporlamayı bildiği bir veritabanında saklamak istediğinizde pratik bir ikili oluşturur. Bu rehberde okumalar için bir tablo oluşturacak, Node-RED'de MQTT topic'lerine abone olacak, her mesajı parametreli sorgularla ekleyecek, mesaj hızı arttığında toplu insert'e geçecek ve tablonun sonsuza kadar büyümemesi için saklama süresini ayarlayacaksınız. Her şey MariaDB'de de aynı şekilde çalışır.

MQTT Verileri Neden MySQL'e Kaydedilmeli?

Bir MQTT broker mesajları taşır; bir veri geçmişi (historian) değildir. Retained message'lar size her topic için en son değeri verir, ancak geçen haftanın değerlerini vermez. Rapor hazırlamak, vardiyaları karşılaştırmak veya bir BI aracını beslemek için geçmişin kalıcı bir yerde durması gerekir.

InfluxDB gibi zaman serisi veritabanları yüksek hızlı telemetri için doğal bir tercihtir; bunu Node-RED ile InfluxDB yazımızda ele alıyoruz. Yine de MySQL ve MariaDB birçok durumda öne çıkar: şirket bunları zaten çalıştırıyordur, raporlama araçları bunlara yerel olarak bağlanır, okumaları iş emirleri veya varlık listeleri gibi ilişkisel verilerle birleştirmeniz gerekir ya da veri hızı makul düzeydedir. Dakikada birkaç yüz okuma, iyi indekslenmiş bir tablo için rahat bir bölgedir.

Adım 1: Veritabanını ve Tabloyu Oluşturun

Henüz bir veritabanı sunucunuz yoksa test için MariaDB'yi Docker'da başlatın:

docker run -d --name iot-mariadb -p 3306:3306 \
  -e MARIADB_ROOT_PASSWORD=change-root \
  -e MARIADB_DATABASE=iot \
  -e MARIADB_USER=nodered \
  -e MARIADB_PASSWORD=change-me \
  mariadb:11

Ardından tabloyu oluşturun. Her okuma için bir satır; izlenebilirlik için orijinal topic, öngörmediğiniz her şey için de tam payload saklanır:

CREATE TABLE mqtt_readings (
  id          BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
  received_at DATETIME(3)  NOT NULL,
  topic       VARCHAR(255) NOT NULL,
  device      VARCHAR(64)  NOT NULL,
  metric      VARCHAR(64)  NOT NULL,
  value       DOUBLE       NULL,
  payload     TEXT         NULL,
  INDEX idx_device_metric_time (device, metric, received_at),
  INDEX idx_time (received_at)
);

Zaman damgalarını UTC olarak saklayın ve dönüşümü raporlarınızda yapın. Bir zaman serisi tablosuna yerel saati ve yaz saati geçişlerini karıştırmak, kafa karıştırıcı boşlukların ve tekrarlanan saatlerin garantili kaynağıdır.

Node-RED kullanıcısına yalnızca ihtiyaç duyduğu yetkileri verin:

GRANT SELECT, INSERT, DELETE ON iot.mqtt_readings TO 'nodered'@'%';

Adım 2: Node-RED MySQL Node'unu Kurun

Menu → Manage palette üzerinden node-red-node-mysql paketini kurun. Bu paket, veritabanı yapılandırmasına sahip bir mysql node'u ekler: host, port (3306), kullanıcı, parola, veritabanı adı ve saat dilimi. Tarihlerin UTC olarak yorumlanması için saat dilimini Z veya +00:00 olarak ayarlayın.

Node basit bir sözleşmeye uyar:

• msg.topic SQL ifadesini taşır.
• msg.payload parametreleri taşır: ? yer tutucuları için bir dizi ya da adlandırılmış :name yer tutucuları için bir nesne.
• Sonuç msg.payload içinde döner: SELECT için satırlar, INSERT için affectedRows ve insertId içeren bir sonuç nesnesi.

Adım 3: MQTT'ye Abone Olun

Broker'ınıza bağlı bir mqtt in node'u ekleyin. Test için broker.connectmq.com adresindeki herkese açık MQTT test broker'ını port 1883, benzersiz bir client ID ile kullanın ve QoS 1, çıktı ayarı auto-detect olacak şekilde connectmq/demo/db91kz/+/+ topic'ine abone olun. İki wildcard seviyesi cihazı ve metriği yakalar, örneğin connectmq/demo/db91kz/boiler1/temperature. db91kz yerine kendi rastgele dizenizi kullanın; herkese açık broker'daki her şey başkaları tarafından görülebilir.

Adım 4: Parametreli Sorgularla Insert

Bu rehberin en önemli kuralı şudur: MQTT payload'larını bir dizeye ekleyerek asla SQL oluşturmayın. Topic'ler ve payload'lar cihazlardan gelir ve paylaşımlı ya da yanlış yapılandırılmış bir broker'da potansiyel olarak herkesten gelebilir. 1); DROP TABLE mqtt_readings; -- gibi bir payload, çalıştırılmak yerine bir sütunda zararsız bir metin olarak kalmalıdır.

mqtt in node'u ile mysql node'u arasına bir function node'u ekleyin:

// Topic: connectmq/demo/db91kz/<device>/<metric>
const mqttTopic = msg.topic;
const parts = mqttTopic.split("/");
if (parts.length !== 5) return null;

const device = parts[3].substring(0, 64);
const metric = parts[4].substring(0, 64);

// Sayıları, sayısal dizeleri veya {"value": ...} nesnelerini kabul et
let raw = msg.payload;
let value = null;
if (typeof raw === "number") {
    value = raw;
} else if (typeof raw === "string" && raw.trim() !== "" && !isNaN(Number(raw))) {
    value = Number(raw);
} else if (typeof raw === "object" && raw !== null && typeof raw.value === "number") {
    value = raw.value;
}

const payloadText = typeof raw === "string" ? raw : JSON.stringify(raw);
// UTC olarak MySQL DATETIME(3) formatı: 2026-10-16 08:15:30.123
const receivedAt = new Date().toISOString().slice(0, 23).replace("T", " ");

msg.topic =
    "INSERT INTO mqtt_readings (received_at, topic, device, metric, value, payload) " +
    "VALUES (?, ?, ?, ?, ?, ?)";
msg.payload = [receivedAt, mqttTopic, device, metric, value, payloadText];
return msg;

msg.topic değerinin SQL ifadesiyle üzerine yazıldığına dikkat edin; kodun topic sütunu için MQTT topic'ini önce mqttTopic içine kaydetmesinin nedeni budur. Function node'unu mysql node'una bağlayın ve her satırın insertId değerini görmek için arkasına bir debug node'u ekleyin.

Insert işlemini test etme

mosquitto_pub -h broker.connectmq.com -p 1883 -i sim-db91kz -t "connectmq/demo/db91kz/boiler1/temperature" -m "71.4"
mosquitto_pub -h broker.connectmq.com -p 1883 -i sim-db91kz -t "connectmq/demo/db91kz/boiler1/pressure" -m '{"value": 2.3, "unit": "bar"}'
mosquitto_pub -h broker.connectmq.com -p 1883 -i sim-db91kz -t "connectmq/demo/db91kz/boiler1/state" -m "running"

Ardından tabloyu kontrol edin:

SELECT received_at, device, metric, value, payload
FROM mqtt_readings
ORDER BY id DESC
LIMIT 10;

Üçüncü mesaj value NULL olarak ve payload sütununda running değeriyle kaydedilir; durumlar ve diğer sayısal olmayan veriler için tam olarak istediğiniz de budur.

Adım 5: Yüksek Mesaj Hızları için Toplu Insert

Saniyede birkaç mesaj için mesaj başına bir INSERT yeterlidir. Bunun ötesinde her gidiş-dönüş ve commit birikir. Toplu yazma, hem Node-RED hem de veritabanı üzerindeki yükü azaltır.

Adım 4'teki function'ı bir SQL ifadesi yerine bir satır dizisi üretecek şekilde değiştirin (msg.payload = [receivedAt, ...] olarak ayarlayın ve msg.topic değerine dokunmayın), ardından manuel modda bir join node'u ekleyin: her msg.payload değerini bir array içinde birleştirin, 100 mesajdan sonra veya 5 saniyelik zaman aşımından sonra gönderin. Sonucun tek düz bir dizi değil, satırlardan oluşan bir dizi (dizilerin dizisi) olduğunu bir debug node'uyla kontrol edin. Arkasından şu function'ı ekleyin:

const rows = msg.payload;            // 6 elemanlı dizilerden oluşan dizi
if (!Array.isArray(rows) || rows.length === 0) return null;

const placeholders = rows.map(() => "(?, ?, ?, ?, ?, ?)").join(", ");
msg.topic =
    "INSERT INTO mqtt_readings (received_at, topic, device, metric, value, payload) VALUES " +
    placeholders;
msg.payload = rows.flat();
return msg;

Her değer hâlâ parametre olarak geçirildiği için toplu yazma, tekli insert kadar güvenlidir. Bir toplu iş beklerken Node-RED durursa en fazla beş saniyelik veri kaybolur; çoğu izleme senaryosu için bu kabul edilebilirdir, daha katı gereksinimler için tekli insert'lerle devam edin.

Adım 6: Saklama Süresi ve İşe Yarar Sorgular

Ham okumaların ne kadar süre saklanacağına karar verin. Node-RED'de basit bir günlük temizlik, belirli bir saatte tekrarlanacak şekilde ayarlanmış bir inject node'u ve ardından bir function node'u kullanır:

msg.topic = "DELETE FROM mqtt_readings WHERE received_at < ? LIMIT 50000";
const cutoff = new Date(Date.now() - 90 * 24 * 3600 * 1000);
msg.payload = [cutoff.toISOString().slice(0, 23).replace("T", " ")];
return msg;

LIMIT, tabloyu uzun süre kilitlememesi için her silme işlemini küçük tutar; bir gün bundan daha fazla satır üretiyorsa işlemi daha sık çalıştırın. Dilerseniz MySQL'in event scheduler'ı aynı işi veritabanının içinde yapabilir.

Sürekli kullanacağınız iki sorgu:

-- Son 24 saat için cihaz başına saatlik ortalama
SELECT device,
       DATE_FORMAT(received_at, '%Y-%m-%d %H:00') AS hour,
       AVG(value) AS avg_value,
       MAX(value) AS max_value
FROM mqtt_readings
WHERE metric = 'temperature'
  AND received_at >= UTC_TIMESTAMP() - INTERVAL 24 HOUR
GROUP BY device, hour
ORDER BY device, hour;

-- 15 dakikadır veri göndermeyen cihazlar
SELECT device, MAX(received_at) AS last_seen
FROM mqtt_readings
GROUP BY device
HAVING last_seen < UTC_TIMESTAMP() - INTERVAL 15 MINUTE;

İkinci sorgu, sessiz kalan cihazları bulmanın ucuz bir yoludur. Bakmak zorunda kalmak yerine bildirim almak için bunu Node-RED e-posta uyarı akışıyla birleştirin. Geçmişi bir panoda göstermek için Node-RED grafik geçmişi rehberi verinin nasıl sorgulanıp çizileceğini anlatır.

Node-RED MySQL Akışlarında Sorun Giderme

Üretim Kurulumu

Gerçek veriler için herkese açık broker'dan ayrılın. ConnectMQ üzerinde özel bir broker oluşturun, ardından Node-RED broker node'unu port 8883, TLS etkin ve portaldaki YOUR_USERNAME, YOUR_PASSWORD ve YOUR_CLIENT_ID değerleriyle yapılandırın. Kısa Node-RED yeniden başlatmalarında broker'ın hâlâ iletebileceği mesajları kaybetmemek için QoS 1 ile kalıcı bir oturum (clean session kapalı) kullanın; ücretsiz Serverless planında QoS 1'e kadar çıkar ve bu zaten kayıt için doğru seviyedir.

Bu sorumluluk ayrımı bilinçlidir. ConnectMQ, mesajları ileten tam yönetilen bir MQTT hizmetidir; veritabanlarına kendisi yazmaz. Veritabanıyla konuşan istemci Node-RED'dir; bu da şemanızı, saklama sürenizi ve SQL'inizi tamamen sizin kontrolünüzde tutar. Kalan broker node ayarları için Node-RED MQTT rehberine göz atın.

MQTT Verilerinizi Node-RED ve ConnectMQ ile Kaydedin

app.connectmq.com adresinde kredi kartı gerektirmeyen ücretsiz Serverless planında bir broker oluşturun, Node-RED'i TLS ile bağlayın ve düz SQL ile sorgulayabileceğiniz bir geçmiş oluşturmaya başlayın. Portalın canlı topic gezgini, tablonuzu tasarlamadan önce topic'leri ve payload formatlarını doğrulamanıza yardımcı olur. Barındırma seçeneklerini hâlâ tartıyorsanız bulut MQTT broker rehberimiz yaklaşımları karşılaştırır.

Sık Sorulan Sorular

Node-RED MySQL node'una parametreleri nasıl geçiririm?

? yer tutuculu SQL'i msg.topic içine, değer dizisini de msg.payload içine koyun. msg.payload bir nesne olduğunda node adlandırılmış :name yer tutucularını da kabul eder.

Bu MariaDB ile çalışır mı?

Evet. node-red-node-mysql hem MySQL'e hem de MariaDB'ye bağlanır; bu rehberdeki tablo, insert'ler ve sorgular her ikisinde de değişiklik yapmadan çalışır.

MQTT verileri için MySQL mi yoksa InfluxDB mi kullanmalıyım?

İlişkisel verilerle join, mevcut raporlama araçları veya orta düzey veri hızları gerektiğinde MySQL ya da MariaDB kullanın. Yüksek hızlı telemetri ve downsampling ile uzun saklama süreleri için bir zaman serisi veritabanı kullanın.

ConnectMQ mesajları doğrudan bir veritabanına kaydedebilir mi?

Hayır. ConnectMQ, veritabanı entegrasyonları olmayan bir MQTT broker'dır. Bu rehberde gösterildiği gibi broker'a bağlanan ve veritabanına yazan Node-RED'i veya başka bir subscriber'ı çalıştırın.