MQTT Veri Historian'ı: Endüstriyel Zaman Serilerini Kaydetmek

Bir MQTT historian, tesisinizin MQTT topic'lerini dinleyen ve her değeri zaman serisi olarak saklayan bir abonedir; böylece operatörler, namespace'in yalnızca o an gösterdiği verilerin trendini çıkarabilir, karşılaştırabilir ve inceleyebilir. Bu rehberde başlıca historian yaklaşımlarını karşılaştıracak, Telegraf'tan InfluxDB'ye ve Python'dan TimescaleDB'ye uzanan iki çalışan veri hattı kuracak ve historian'ınızın bir yıl sonra güvenilir olup olmayacağını belirleyen tasarım kararlarını öğreneceksiniz.

Bir MQTT Historian Ne Yapar?

Klasik bir proses historian'ı SCADA'dan veya PLC'lerden etiketleri sorgular ve bunları tescilli bir arşivde sıkıştırır. MQTT historian ise akışı tersine çevirir: cihazlar ve gateway'ler bir broker'a yayın yapar, historian da yalnızca bir abone daha olur. Bunun faydalı sonuçları vardır:

• Tesise ek yük yok. Historian hiçbir zaman bir PLC'yi sorgulamaz; zaten yayınlanmış olanı okur.
• Tek entegrasyon noktası. Abonelik ilgili dalı kapsıyorsa namespace'e bir makine eklemek onu otomatik olarak historian'a da ekler.
• Topic'lerden gelen bağlam. ISA-95 topic'leriyle saha, alan, hat ve makine, manuel eşleme olmadan etiket veya sütun haline gelir.

Historian'ın üstlenmesi gereken, broker'ın yapmadığı her şeydir: uzun vadeli depolama, verimli zaman aralığı sorguları, örnek azaltma (downsampling) ve saklama. Bir broker son retained değeri ve kısa ömürlü oturum kuyruklarını tutar; aylarca geçmişi değil. Bu iş bölümü unified namespace ve veri gölü yazısında daha ayrıntılı açıklanıyor.

Historian Yaklaşımlarının Karşılaştırması

Tek bir doğru cevap yoktur. Pek çok tesis operatörler ve mühendisler için haftalar veya aylar boyunca yüksek çözünürlüklü veri tutan sıcak bir historian bulundurur ve uzun vadeli analitik için her şeyi bir veri gölüne aktarır.

Test İçin Örnek Veri

İki veri hattı da JSON payload'lı <prefix>/<site>/<area>/<line>/<machine>/telemetry biçiminde topic'ler bekler. Unified namespace nasıl kurulur yazısındaki gateway simülatörü elinizdeyse onu çalıştırın. Aksi halde bu döngü herkese açık MQTT test broker'ına simüle veri yayınlar. Herkese açık alan herkes tarafından okunabilir; bu yüzden kendi rastgele önekinizi ve yalnızca simüle değerler kullanın:

PREFIX="connectmq/demo/h6n3"
while true; do
  TS=$(date -u +%Y-%m-%dT%H:%M:%SZ)
  TEMP=$(awk -v r=$RANDOM 'BEGIN{printf "%.1f", 38 + r % 60 / 10}')
  mosquitto_pub -h broker.connectmq.com -p 1883 -i hist-sim-h6n3 \
    -t "$PREFIX/dallas/packaging/line2/filler/telemetry" \
    -m "{\"ts\":\"$TS\",\"temp_c\":$TEMP,\"speed_bpm\":118}"
  sleep 5
done

Veri Hattı 1: Telegraf ve InfluxDB

Telegraf'ın MQTT consumer eklentisi broker'a abone olur ve hiç özel kod yazmadan InfluxDB'ye yazar. InfluxDB 2'yi Docker'da bir geliştirme token'ıyla başlatın:

docker run -d --name influxdb -p 8086:8086 \
  -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=plant \
  -e DOCKER_INFLUXDB_INIT_BUCKET=historian \
  -e DOCKER_INFLUXDB_INIT_ADMIN_TOKEN=dev-token-change-me \
  influxdb:2.7

Ardından telegraf.conf dosyasını oluşturun. Topic ayrıştırma, her ISA-95 seviyesini bir InfluxDB etiketine dönüştürür:

[agent]
  interval = "10s"
  flush_interval = "10s"

[[inputs.mqtt_consumer]]
  servers = ["tcp://broker.connectmq.com:1883"]
  topics = ["connectmq/demo/h6n3/+/+/+/+/telemetry"]
  qos = 1
  client_id = "telegraf-historian-h6n3"
  name_override = "telemetry"
  data_format = "json"
  json_time_key = "ts"
  json_time_format = "2006-01-02T15:04:05Z07:00"

  [[inputs.mqtt_consumer.topic_parsing]]
    topic = "connectmq/demo/h6n3/+/+/+/+/telemetry"
    tags = "_/_/_/site/area/line/machine/_"

[[outputs.influxdb_v2]]
  urls = ["http://localhost:8086"]
  token = "dev-token-change-me"
  organization = "plant"
  bucket = "historian"

telegraf --config telegraf.conf ile çalıştırın. 8086 portunda InfluxDB arayüzünü açın, historian bucket'ına göz atın; site, area, line ve machine etiketlerine ve temp_c gibi sayısal alanlara sahip bir telemetry ölçümü göreceksiniz. JSON ayrıştırıcı varsayılan olarak sayısal alanları tutar; metin alanlarına ihtiyacınız varsa bunları json_string_fields ile açıkça listeleyin. Telegraf MQTT consumer belgeleri tüm seçenekleri listeliyor.

Özel bir ConnectMQ broker'ı için sunucuyu ssl://YOUR_BROKER_HOST:8883 olarak değiştirin, username = "YOUR_USERNAME" ve password = "YOUR_PASSWORD" ekleyin ve client_id değerini YOUR_CLIENT_ID yapın; çünkü client ID'ler portalda verilen kimlik bilgilerine bağlıdır. Topic'leriniz kuruluş adınızla başladığında demo önekini ve ona ait ayrıştırma seviyelerini kaldırın.

Veri Hattı 2: Python ve TimescaleDB

SQL ve üretim siparişleri ya da parti kayıtlarıyla join yapmak istediğinizde TimescaleDB doğal bir seçimdir. Docker'da başlatın:

docker run -d --name timescaledb -p 5432:5432 -e POSTGRES_PASSWORD=historian timescale/timescaledb:latest-pg16

Aşağıdaki abone veriyi uzun formatta, metrik başına bir satır olarak saklar; bu da farklı alanlar yayınlayan makinelerle iyi başa çıkar. Bağımlılıkları pip install "paho-mqtt>=2.0" "psycopg[binary]" ile kurun.

import json
import os
from datetime import datetime, timezone

import paho.mqtt.client as mqtt
import psycopg

DSN = os.getenv("PG_DSN", "postgresql://postgres:historian@localhost:5432/postgres")
PREFIX = os.getenv("UNS_PREFIX", "connectmq/demo/h6n3")

conn = psycopg.connect(DSN, autocommit=True)
conn.execute("""
    CREATE TABLE IF NOT EXISTS telemetry (
        time    TIMESTAMPTZ NOT NULL,
        site    TEXT, area TEXT, line TEXT, machine TEXT,
        metric  TEXT NOT NULL,
        value   DOUBLE PRECISION
    )""")
conn.execute("SELECT create_hypertable('telemetry', 'time', if_not_exists => TRUE)")

def on_connect(client, userdata, flags, reason_code, properties):
    print("historian connected:", reason_code)
    client.subscribe(f"{PREFIX}/+/+/+/+/telemetry", qos=1)

def on_message(client, userdata, msg):
    if msg.retain:
        return  # retained değerler ilk yayınlandıklarında zaten kaydedildi
    site, area, line, machine, _ = msg.topic[len(PREFIX) + 1:].split("/")
    try:
        data = json.loads(msg.payload)
    except ValueError:
        return
    ts = data.pop("ts", None)
    when = datetime.fromisoformat(ts.replace("Z", "+00:00")) if ts else datetime.now(timezone.utc)
    rows = [(when, site, area, line, machine, k, float(v))
            for k, v in data.items() if isinstance(v, (int, float)) and not isinstance(v, bool)]
    with conn.cursor() as cur:
        cur.executemany(
            "INSERT INTO telemetry (time, site, area, line, machine, metric, value) "
            "VALUES (%s, %s, %s, %s, %s, %s, %s)", rows)

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="pg-historian-h6n3")
client.on_connect = on_connect
client.on_message = on_message
client.connect(os.getenv("MQTT_HOST", "broker.connectmq.com"), 1883, keepalive=60)
client.loop_forever()

Standart SQL ve Timescale'in time_bucket fonksiyonuyla makine başına beş dakikalık ortalamaları sorgulayın:

SELECT time_bucket('5 minutes', time) AS bucket, line, machine, avg(value) AS avg_temp
FROM telemetry
WHERE metric = 'temp_c' AND time > now() - interval '1 day'
GROUP BY bucket, line, machine
ORDER BY bucket;

Neyin Saklanacağına Karar Vermek

Yayınlanan her değeri sonsuza kadar saklamak güvenli görünür, ancak sorguları yavaşlatır ve maliyetler sessizce büyür. Historian devreye girmeden önce operasyon ve mühendislik ekipleriyle bir depolama politikası üzerinde uzlaşın:

• Proses değerleri (sıcaklık, basınç, hız gibi): birkaç hafta boyunca yayınlanma hızında ham olarak saklayın, ardından bir dakikalık minimum, maksimum ve ortalama gibi özetleri tutun.
• Sayaçlar (iyi ve hurda sayıları gibi): her değişikliği saklayın; çünkü OEE ve verim için önemli olan okumalar arasındaki farklardır.
• Durumlar ve alarmlar: her geçişi zaman damgasıyla saklayın. Hacimleri düşüktür ve duruş analizi için vazgeçilmezdir.
• Olaylar (parti tamamlanmaları gibi): hepsini saklayın; ideal olarak genel metrikler yerine kendi alanlarına sahip bir tabloda.
• Tanılama verileri (gateway çalışma süresi veya sinyal gücü gibi): belirli bir sorunu araştırmıyorsanız daha kısa bir saklama süresi uygulayın.

Bunu topic standardınızda her ad alanı için yazmak, historian yapılandırmasının gelişigüzel büyümek yerine namespace tasarımını izlemesini sağlar.

Önemli Tasarım Kararları

1. Varış zamanı yerine kaynak zaman damgası. Tamponlanmış veya gecikmiş verinin doğru ana yerleşmesi için payload'daki zaman damgasını saklayın. Gecikmeyi ölçmeniz gerekiyorsa varış zamanını da tutun.
2. Kalıcı oturumlar. Historian'a sabit bir client ID ve kalıcı bir oturum verin (clean_session=False veya MQTT 5 oturum sona erme süresi); böylece broker kısa bir yeniden başlatma sırasında QoS 1 mesajlarını kuyrukta tutar.
3. Uzun kesintiler için sahada tamponlama. Broker kuyrukları kısa boşlukları karşılar. Saatler süren WAN kesintileri için gateway'de tamponlayın; bkz. MQTT ile store and forward.
4. Retained message yönetimi. Abonelik sırasında broker retained değerleri yeniden gönderir. Yeniden başlatmaların yinelenen satırlar oluşturmaması için bunları atlayın veya zaman ve metriğe göre tekilleştirin.
5. Ölü bantlar ve istisna bazlı raporlama. Yalnızca anlamlı değişiklikleri yayınlamak, bilgi kaybetmeden depolamayı ciddi ölçüde azaltır.
6. Saklama ve örnek azaltma. Ham veriyi belirli bir süre, sürekli özetleri veya azaltılmış örnek kovalarını ise daha uzun süre tutun.
7. Yatay ölçekleme. MQTT 5 paylaşımlı abonelikleriyle birden fazla historian çalışanı $share/historian/... aboneliğini paylaşabilir; böylece her mesaj bir kez yazılır. ConnectMQ'da paylaşımlı abonelikler ücretli Serverless plandan itibaren sunulur.

ConnectMQ yerleşik bir veritabanı veya historian bağlayıcısı içermez ve bu desen için buna ihtiyaç da duymaz: yukarıdaki abonelerin her biri normal bir MQTT istemcisi olarak bağlanır. Verinin kendisini modellemek hakkında daha fazlası için endüstriyel zaman serisi verisi yazısını okuyun; daha geniş mimari için endüstriyel IoT MQTT rehberine bakın.

MQTT Historian'ınızı ConnectMQ'dan Besleyin

Bir historian, onu besleyen akış kadar eksiksizdir. ConnectMQ; MQTT 3.1.1 ve 5.0, QoS 1 ve 2, kalıcı oturumlar, TLS ve ISA-95 farkındalığına sahip bir portal sunan tam yönetilen bir MQTT hizmetidir; böylece Telegraf veya Python historian'ınız güvenilir bir kaynağa sahip olur. app.connectmq.com adresinde kredi kartı gerektirmeden ücretsiz Serverless planla bir broker oluşturun veya tesisiniz için bir bulut MQTT broker çalıştırmak hakkında daha fazla bilgi edinin.

Sık Sorulan Sorular

Bir MQTT broker historian mıdır?

Hayır. Broker mesajları yönlendirir ve her topic için son retained değeri tutar. Historian ise broker'a abone olur ve sorgulama için zaman serisinin tamamını saklar.

InfluxDB mi yoksa TimescaleDB mi seçmeliyim?

En hızlı, yalnızca yapılandırmaya dayalı kurulum ve Grafana panoları için Telegraf ile InfluxDB'yi seçin. SQL ve siparişler ile partiler gibi ilişkisel verilerle join istediğinizde TimescaleDB'yi seçin.

Ticari bir historian MQTT'den okuyabilir mi?

Pek çok ticari historian MQTT veya Sparkplug bağlayıcıları sunar. Bağlayıcının payload formatınızı ve topic yapınızı destekleyip desteklemediğini, kullanıyorsanız Sparkplug birth mesajlarını ve alias'ları işleyip işleyemediğini kontrol edin.

Historian yeniden başladığında veri kaybını nasıl önlerim?

Broker'ın mesajları kısa süreliğine kuyrukta tutması için kalıcı oturum ve QoS 1 ile sabit bir client ID kullanın; daha uzun kesintiler için sahada tamponlama yapın.