Node-RED InfluxDB Eğitimi: MQTT Verisini Saklama ve Sorgulama

Node-RED ve InfluxDB birlikte size küçük ama eksiksiz bir telemetri hattı sunar: MQTT mesajları gelir, Node-RED onları şekillendirir ve InfluxDB onları sorgulayıp grafiğe dökebileceğiniz zaman serileri olarak saklar. Bu Node-RED InfluxDB eğitiminde InfluxDB 2'yi Docker'da çalıştıracak, MQTT okumalarını doğru field ve tag'lerle yazacak ve bir grafik çizmek için onları Flux ile geri sorgulayacaksınız. Her adım ücretsiz bir broker kullanır; böylece tüm zinciri yarım saatten kısa sürede test edebilirsiniz.

Mimari: MQTT Girer, Zaman Serisi Çıkar

Hattın üç rolü vardır ve bunları ayrı tutmak daha sonra ölçeklendirmeyi kolaylaştırır:

1. Cihazlar JSON okumalarını connectmq/demo/a7f3k2/room1/telemetry gibi MQTT topic'lerine yayınlar.
2. Broker bu mesajları her aboneye yönlendirir. Geçmiş saklamaz; bu veritabanının işidir. Bu ayrım sizin için yeniyse, MQTT broker'ın ne yaptığını ayrıntılı olarak okuyabilirsiniz.
3. Node-RED abone olur, her mesajı doğrular ve bir InfluxDB noktasına dönüştürüp yazar. İkinci bir dal, grafikler için InfluxDB'yi sorgular.

Aynı yerde yönlendirme, uyarılar ya da bir pano da istiyorsanız Node-RED iyi bir seçimdir. Yalnızca yüksek hacimde veri alımına ihtiyacınız varsa, Telegraf'ın MQTT tüketicisi bir alternatiftir; bu, MQTT verisini InfluxDB'de saklama rehberinde ele alınıyor.

Adım 1: InfluxDB 2'yi Docker ile Çalıştırın

Resmi imaj ilk başlatmada bir organizasyon, bucket ve yönetici token'ı oluşturabilir; bu da birkaç tıklamadan tasarruf sağlar:

docker run -d --name influxdb -p 8086:8086 \
  -v influxdb-data:/var/lib/influxdb2 \
  -e DOCKER_INFLUXDB_INIT_MODE=setup \
  -e DOCKER_INFLUXDB_INIT_USERNAME=admin \
  -e DOCKER_INFLUXDB_INIT_PASSWORD=change-me-please \
  -e DOCKER_INFLUXDB_INIT_ORG=demo-org \
  -e DOCKER_INFLUXDB_INIT_BUCKET=sensors \
  -e DOCKER_INFLUXDB_INIT_RETENTION=30d \
  -e DOCKER_INFLUXDB_INIT_ADMIN_TOKEN=my-super-secret-token \
  influxdb:2.7

http://localhost:8086 adresini açın, oturum açın; 30 günlük saklama süresine sahip sensors bucket'ını göreceksiniz. Üretim için Node-RED'e yönetici token'ını vermek yerine yalnızca o bucket'a yazma erişimi olan ayrı bir API token'ı oluşturun. Bu eğitim InfluxDB 2.x'i hedefler; InfluxDB 3 sorgu tarafını değiştirir, bu yüzden Flux örneklerini kopyalamadan önce uyumluluk notlarını kontrol edin.

Adım 2: MQTT'ye Test Verisi Yayınlayın

Bu betik iki odayı simüle eder ve ücretsiz herkese açık MQTT broker'ı olan broker.connectmq.com adresine yayın yapar. a7f3k2 değerini kendi rastgele dizenizle değiştirin; herkese açık broker'daki topic'ler herkese görünür.

# pip install "paho-mqtt>=2.0"
import json
import random
import time
import uuid

import paho.mqtt.client as mqtt

PREFIX = "connectmq/demo/a7f3k2"
ROOMS = {"room1": 22.0, "room2": 25.0}

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                     client_id=f"influx-sim-{uuid.uuid4().hex[:8]}")
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_start()

try:
    while True:
        for room in ROOMS:
            ROOMS[room] += random.uniform(-0.2, 0.2)
            payload = {
                "temperature": round(ROOMS[room], 2),
                "humidity": round(random.uniform(38, 60), 1),
                "ts": int(time.time() * 1000),
            }
            client.publish(f"{PREFIX}/{room}/telemetry", json.dumps(payload), qos=1)
        time.sleep(5)
except KeyboardInterrupt:
    pass
finally:
    client.loop_stop()
    client.disconnect()

Bir cihaz zaman damgası (ts) eklemek iyi bir uygulamadır: mesajlar gecikirse veritabanı yine de okumanın ne zaman geldiğini değil, ne zaman alındığını kaydeder.

Adım 3: node-red-contrib-influxdb Paketini Kurun

Node-RED'de Manage palette → Install yolunu açın ve node-red-contrib-influxdb paketini kurun. Üç node elde edersiniz: tekil yazmalar için influxdb out, birçok noktayı bir kerede yazmak için influxdb batch ve sorgular için influxdb in. Hepsi bir influxdb yapılandırma node'unu paylaşır.

Yapılandırma node'unu ayarlayın:

Adım 4: MQTT Payload'larını Field ve Tag'lere Eşleyin

InfluxDB field'ları (ölçülen değerler, indekslenmez) tag'lerden (filtreleme ve gruplama yaptığınız indeksli meta veri) ayırır. Bunu doğru yapmak, tüm Node-RED InfluxDB kurulumundaki en önemli tasarım kararıdır:

• Sıcaklık ve nem gibi sayısal okumaları field'lara koyun.
• Oda, hat ya da cihaz kimliği gibi düşük kardinaliteli tanımlayıcıları tag'lere koyun.
• Sınırsız değerleri (zaman damgaları, rastgele kimlikler, serbest metin) asla tag'lere koymayın; her benzersiz tag kombinasyonu yeni bir seri oluşturur ve bellek kullanımı onunla birlikte büyür.

connectmq/demo/a7f3k2/+/telemetry topic'ine abone olan, çıktısı "a parsed JSON object" olan bir mqtt in node'u, ardından bu function node'unu ekleyin. influxdb out node'u iki nesneden oluşan bir dizi kabul eder: önce field'lar, sonra tag'ler.

// Function node "to influx point"
const p = msg.payload;
if (typeof p !== "object" || p === null) {
    return null;
}

const fields = {};
if (typeof p.temperature === "number") fields.temperature = p.temperature;
if (typeof p.humidity === "number") fields.humidity = p.humidity;
if (Object.keys(fields).length === 0) {
    node.warn("No numeric fields in " + msg.topic);
    return null;
}

// connectmq/demo/a7f3k2/room1/telemetry -> room1
const room = msg.topic.split("/")[3];

// Varsa cihaz zaman damgasını kullan (milisaniye)
if (typeof p.ts === "number") {
    fields.time = new Date(p.ts);
}

msg.payload = [fields, { room: room, source: "node-red" }];
msg.measurement = "environment";
return msg;

Onu organizasyonu demo-org, bucket'ı sensors ve zaman hassasiyeti milisaniye olarak ayarlanmış bir influxdb out node'una bağlayın. msg.measurement kullanılsın diye node'da measurement alanını boş bırakın. Deploy edin, bir dakika bekleyin ve InfluxDB arayüzünde Data Explorer'ı açın: temperature ve humidity field'larına ve bir room tag'ine sahip environment measurement'ı görünür.

Node sürümünüz field nesnesi içinde bir time alanını kabul etmiyorsa o iki satırı kaldırın; InfluxDB bu durumda yazma zamanını kullanır.

Adım 5: Flux ile Sorgulayın ve Sonucu Grafiğe Dökün

Sorgulamak da aynı derecede basittir. Bir inject node'u her dakika, Flux sorgusunu oluşturan bir function node'unu tetikler:

// Function node "build flux query"
msg.query = `
from(bucket: "sensors")
  |> range(start: -6h)
  |> filter(fn: (r) => r._measurement == "environment" and r._field == "temperature")
  |> aggregateWindow(every: 1m, fn: mean, createEmpty: false)
  |> keep(columns: ["_time", "_value", "room"])
`;
return msg;

Onu organizasyonu demo-org olan bir influxdb in node'una gönderin. Node, her biri _time, _value ve tag sütunlarını içeren bir satır dizisi döndürür. Bunları oda başına bir seri olacak şekilde grafik noktalarına dönüştürün:

// Function node "rows to chart"
const rows = Array.isArray(msg.payload) ? msg.payload : [];
const byRoom = {};

for (const r of rows) {
    if (!byRoom[r.room]) byRoom[r.room] = [];
    byRoom[r.room].push({ x: Date.parse(r._time), y: Math.round(r._value * 100) / 100 });
}

const out = [{ payload: [] }];   // yeniden çizmeden önce grafiği temizle
for (const room of Object.keys(byRoom)) {
    out.push({ topic: room, payload: byRoom[room] });
}
return [out];

Bunu serileri msg.topic değerinden, X anahtarı x ve Y anahtarı y olan bir Dashboard 2.0 ui-chart node'una bağlayın. Veri artık veritabanından geldiği için grafik yeniden başlatmalardan otomatik olarak sağ çıkar. InfluxDB ihtiyacınızdan fazlaysa Node-RED grafik geçmişi rehberi bunu daha hafif seçeneklerle karşılaştırıyor; Dashboard 2.0 eğitimi ise grafik ayarlarını açıklıyor.

Üretim Kontrol Listesi

1. Yazıcı akış için tek bir bucket'a yazma erişimi, sorgu akışı için okuma erişimi olan özel bir InfluxDB token'ı kullanın.
2. Ham veriye ne kadar süre ihtiyaç duyduğunuza uyan bir bucket saklama politikası belirleyin; uzun trendlere ihtiyacınız varsa eski veriyi bir InfluxDB task'ı ile ikinci bir bucket'a indirgeyin.
3. Tag'leri düşük kardinaliteli ve cihazlar arasında tutarlı tutun.
4. Function node'unun yaptığı gibi yazmadan önce payload'ları doğrulayın; böylece tek bir hatalı cihaz bucket'ı dizelerle dolduramaz.
5. Kısa Node-RED yeniden başlatmaları okuma kaybettirmesin diye abonelikte QoS 1 ve kalıcı bir oturum kullanın.
6. Saniyede yüzlerce nokta yazıyorsanız, mesajları önce bir join node'uyla gruplayarak influxdb batch node'unu kullanın.
7. Yazma hatalarını izleyin: influxdb out node'una kapsamlanmış bir catch node'u ekleyin ve hataları bir debug node'una ya da uyarıya yönlendirin.

Sorun Giderme

Field türü çakışması dikkat gerektirir: InfluxDB, bir field'ın türünü bir shard içindeki ilk yazmada sabitler. Bir cihaz 22, sonra 22.5 gönderirse, tamsayı ile ondalık uyuşmazlığı reddedilir. JSON bunları ayırt etmediği için tam sayılar gönderen sensörler zaman zaman bunu tetikler. Değerleri firmware'den tutarlı şekilde göndermek (örneğin her zaman bir ondalık basamakla) ve ilk yazmalardan sonra Data Explorer'da field türünü kontrol etmek, ileride sürprizleri önler.

MQTT Tarafını da Güvenli Hale Getirin

Herkes veri aldığı topic'lere sahte okumalar yayınlayabiliyorsa, özenle ayarlanmış token'lara sahip bir veritabanı boşa gider. Herkese açık broker'da durum tam olarak budur; bu yüzden yalnızca test içindir. Üretim için Node-RED mqtt-broker yapılandırma node'unu 8883 portunda TLS ve kimlik bilgileri kullanan özel bir broker'a geçirin. ConnectMQ'da her broker üretilmiş bir kullanıcı adı, parola ve client ID alır ve client ID'ler bu kimlik bilgilerine bağlıdır; böylece rastgele bir istemci veri alım akışınızın kimliğine bürünemez.

Bir bulut MQTT broker ağı da basitleştirir: sahadaki cihazlar ve InfluxDB'nin yanındaki Node-RED, broker'a dışarı doğru bağlanır; böylece veritabanı ağına port açmanız gerekmez. ConnectMQ'nun kendisi veritabanlarına yazmaz; dürüst desen tam olarak bu makaledeki gibidir: InfluxDB'ye yazan bir Node-RED (ya da Telegraf) abonesi. Broker'lar, joker karakterler ve yapılandırma node'ları hakkında temel bilgiler için Node-RED MQTT eğitimine bakın.

MQTT Verinizi ConnectMQ ile Saklamaya Başlayın

app.connectmq.com adresinde kredi kartı gerektirmeden Serverless planında ücretsiz bir hesap oluşturun. Bir broker oluşturun, kimlik bilgilerini Node-RED yapılandırma node'unuza kopyalayın; InfluxDB hattınız, portalın topic gezgininde canlı olarak inceleyebileceğiniz özel, TLS ile şifrelenmiş bir akışla beslensin.

Sık Sorulan Sorular

node-red-contrib-influxdb hangi InfluxDB sürümlerini destekler?

InfluxQL ile InfluxDB 1.x'i, Flux ile 1.8'i ve token, organizasyon ve bucket'larla 2.x'i destekler. Node'ların doğru seçenekleri göstermesi için sürümü yapılandırma node'unda seçin.

Her MQTT mesajını InfluxDB'ye yazmalı mıyım?

İleride ihtiyaç duyabileceğiniz her okumayı yazın, ancak uzun vadeli trendler için yüksek frekanslı veriyi ayrı bir bucket'ta indirgeyin. Cihazlar ihtiyacınızdan hızlı yayın yapıyorsa, yazmadan önce Node-RED'de ortalama alın.

InfluxDB'de field ve tag arasındaki fark nedir?

Field'lar ölçülen değerleri saklar ve indekslenmez. Tag'ler oda ya da cihaz kimliği gibi meta verileri saklar, indekslenir ve sınırlı sayıda farklı değere sahip olmalıdır.

Node-RED grafiği yerine Grafana kullanabilir miyim?

Evet. Grafana'yı aynı InfluxDB bucket'ına yönlendirin ve bu rehberdeki Flux sorgusunu kullanın. Node-RED MQTT veri alımını yapmaya devam ederken Grafana görselleştirmeyi üstlenir.