Endüstriyel DataOps: Tesis Verisini Bağlamlandırmak ve Yönetmek

Endüstriyel DataOps, ham operasyonel teknoloji (OT) verisini bağlamlandırılmış, güvenilir bilgiye dönüştürme ve bunu ihtiyaç duyan her tüketiciye tek seferlik bir entegrasyon projesi olarak değil, sürekli olarak ulaştırma disiplinidir. Bu yazıda bir DataOps katmanının üstlendiği dört görevi, MQTT tabanlı bir mimaride nerede durduğunu ve temiz veriyi ortak bir namespace'e yeniden yayınlayan küçük bir bağlamlandırma servisini nasıl kuracağınızı öğreneceksiniz.

Endüstriyel DataOps Aslında Ne Anlama Geliyor?

Terim, DevOps fikirlerini (otomasyon, sürümleme, izleme) veri hatlarına uygulayan BT DataOps'tan ödünç alınmıştır. Tesiste ise sorun farklıdır. Veri; analitik için değil, kontrol için tasarlanmış PLC'ler, sensörler ve makineler tarafından üretilir. Etiketlerde birim yoktur, zaman damgaları farklı saatlerden gelir, aynı makineler farklı adlar kullanır ve aralarındaki eşlemenin sahibi yoktur.

Bir DataOps katmanı olmadan her tüketici bu sorunları kendi başına çözer. Historian ekibi bir eşleme yazar, MES entegratörü bir başkasını, veri bilimi ekibi de bir CSV dışa aktarımını elle temizler. Her düzeltme yereldir ve kırılgandır. Endüstriyel DataOps bu işi kaynaklar ile tüketiciler arasındaki tek bir yönetilen katmana taşır; böylece temizlik bir kez yapılır ve her yerde yeniden kullanılır.

Pratik sonuç, bir değerin varlığı, birimi, zaman damgası, kalitesi ve anlamı eklenmiş olarak geldiği ve bu katmandaki değişikliklerin kod gibi gözden geçirilip sürümlendiği bir tesis veri katmanıdır.

Bir Endüstriyel DataOps Katmanının Dört Görevi

Bağlamlandırma

Bir sayıyı bilgiye dönüştüren şey bağlamdır. Bir sıcaklık okumasının hangi makineye, makinenin hangi parçasına ait olduğunu, hangi ürünün çalıştığını ve hangi vardiyanın görevde olduğunu bilmesi gerekir. Bağlamın bir kısmı statiktir (varlık, konum) ve topic yoluna veya varlık modeline aittir. Bir kısmı ise dinamiktir (güncel sipariş, reçete, çalışma modu) ve en iyi, tüketicilerin telemetriyle birleştirebileceği ayrı durum topic'leri olarak yayınlanır. IIoT veri modelleme rehberi, bu bağlamın tiplere ve hiyerarşilere nasıl yapılandırılacağını ele alıyor.

Normalizasyon

Normalizasyon, benzer şeylerin aynı görünmesini sağlar. Farklı tedarikçilerden gelen iki motor da current değerini amper cinsinden raporlamalı, state değerini aynı numaralandırmayla bildirmeli ve milisaniye epoch zaman damgaları kullanmalıdır. Tipik normalizasyon adımları; hamdan mühendislik birimine ölçekleme, birim dönüştürme, yeniden adlandırma, enum eşleme (tedarikçi durum kodlarından ortak bir kümeye) ve zaman damgalarının UTC'ye hizalanmasıdır.

Doğrulama

OT verisi belirli biçimlerde bozulur: sensörler donar, bir kablo koptuğunda değerler analog aralığın sınırlarına sıçrar, saatler kayar, gateway'ler tamponlanmış veriyi sırasız yeniden gönderir. Bir DataOps katmanı bunları tüketiciler görmeden yakalamalıdır. Şüpheli veriyi silmeniz gerekmez; çoğu zaman onu tutup bir kalite bayrağı eklemek daha iyidir; böylece bir historian onu saklayabilirken bir model eğitici onu filtreleyebilir.

Yönetişim

Yönetişim en az teknik olan ve en sık atlanan görevdir. Birinin varlık tiplerinin sahibi olması, topic yollarındaki değişiklikleri onaylaması ve hangi uygulamaların hangi veriye bağımlı olduğunu bilmesi gerekir. Pratikte bu; modelleri ve eşlemeleri sürüm kontrolünde tutmak, değişiklikleri gözden geçirmek ve tüketicilerin okuyabileceği bir değişiklik günlüğü yayınlamak demektir.

DataOps, MQTT Mimarisinde Nerede Durur?

MQTT tabanlı bir tesiste broker dağıtım omurgasıdır ve DataOps mantığı onun etrafındaki istemcilerde yaşar. Yaygın bir düzen şöyledir:

PLCs / sensors
     |
Edge gateway (protocol conversion, raw tags)
     |  publishes raw/...
MQTT broker  <---->  DataOps services (context, normalize, validate)
     |  republish to modeled namespace: site/area/line/cell/...
Consumers: historian, MES, dashboards, ML pipelines

Ham namespace sorun giderme için faydalıdır, ancak tüketiciler yalnızca modellenmiş namespace'e abone olmalıdır. Bu ayrım, gateway'e dokunmadan bir eşlemeyi düzeltmenize ve eşlemeye dokunmadan tüketici eklemenize olanak tanır. Bu, unified namespace ile aynı ilkedir: güncel durumu yayınlamak ve keşfetmek için tesis hiyerarşisine göre yapılandırılmış tek bir yer.

Araçlar konusunda dürüst bir not: broker mesajları yönlendirir; dönüşüm kodunuzu çalıştırmaz. Bir ConnectMQ kurulumunda bağlamlandırma servisi, ham topic'lere abone olup modellenmiş topic'ler yayınlayan sıradan bir MQTT istemcisidir (bir Python süreci, bir Node-RED akışı veya ticari bir DataOps ürünü). ConnectMQ; yönetilen MQTT broker, ISA-95 topic algılama ve Sparkplug B'yi çözüp ISA-95 topic'lerine yeniden yayınlama özelliklerini sunar; bu da yaygın bir normalizasyon adımını özel kod yazmadan karşılar.

Küçük Bir Bağlamlandırma Servisi Kurmak

Aşağıdaki Python servisi ham etiket mesajlarına abone olur, bir eşleme dosyası uygular, değerleri ölçekler ve doğrular, ardından bunları bir ISA-95 namespace'ine yeniden yayınlar. Deneme amacıyla herkese açık MQTT test broker'ı üzerinde çalışır; orada yalnızca simüle veri kullanın.

Önce sürüm kontrolünde yaşayacak eşleme dosyası:

# mapping.yaml
modelVersion: 1.0.3
tags:
  PLC3.DB10.DBW4:
    path: berlin/line2/filler1/speed
    unit: bpm
    rawMin: 0
    rawMax: 27648
    engMin: 0
    engMax: 200
    validMin: 0
    validMax: 180
  PLC3.DB10.DBW6:
    path: berlin/line2/filler1/temperature
    unit: degC
    rawMin: 0
    rawMax: 27648
    engMin: -20
    engMax: 120
    validMin: 5
    validMax: 90

Ardından servis:

import json
import sys
import time
import uuid

import paho.mqtt.client as mqtt
import yaml

BROKER = "broker.connectmq.com"
PREFIX = sys.argv[1] if len(sys.argv) > 1 else "connectmq/demo/" + uuid.uuid4().hex[:8]
STALE_SECONDS = 600

with open("mapping.yaml", encoding="utf-8") as f:
    MAPPING = yaml.safe_load(f)

last_seen = {}  # etiket -> (ham değer, bu değerin ilk görüldüğü zaman)


def scale(raw, m):
    span = m["rawMax"] - m["rawMin"]
    return m["engMin"] + (raw - m["rawMin"]) * (m["engMax"] - m["engMin"]) / span


def quality(tag, raw, value, m, now):
    prev = last_seen.get(tag)
    if prev is None or prev[0] != raw:
        last_seen[tag] = (raw, now)
    elif now - prev[1] > STALE_SECONDS:
        return "STALE"
    if not (m["validMin"] <= value <= m["validMax"]):
        return "OUT_OF_RANGE"
    return "GOOD"


def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(f"{PREFIX}/raw/#", qos=1)
    print("listening on", f"{PREFIX}/raw/#")


def on_message(client, userdata, msg):
    tag = msg.topic.split("/raw/", 1)[1]
    m = MAPPING["tags"].get(tag)
    if m is None:
        client.publish(f"{PREFIX}/dataops/unmapped", json.dumps({"tag": tag}), qos=0)
        return
    raw = json.loads(msg.payload)
    now = time.time()
    value = round(scale(raw["v"], m), 2)
    out = {
        "value": value,
        "unit": m["unit"],
        "ts": raw.get("ts", int(now * 1000)),
        "quality": quality(tag, raw["v"], value, m, now),
        "modelVersion": MAPPING["modelVersion"],
    }
    client.publish(f"{PREFIX}/{m['path']}", json.dumps(out), qos=1)


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="dataops-" + uuid.uuid4().hex[:8])
client.on_connect = on_connect
client.on_message = on_message
client.connect(BROKER, 1883, keepalive=60)
client.loop_forever()

Bağımlılıkları pip install paho-mqtt pyyaml ile kurun, servisi seçtiğiniz bir önekle başlatın ve ona ham bir değer gönderin:

python dataops.py connectmq/demo/plant42
mosquitto_pub -h broker.connectmq.com -p 1883 -i raw-test-42 \
  -t "connectmq/demo/plant42/raw/PLC3.DB10.DBW4" -m '{"v": 16000, "ts": 1760000000000}'
mosquitto_sub -h broker.connectmq.com -p 1883 -i sub-test-42 -t "connectmq/demo/plant42/berlin/#" -v

Abone; birim, zaman damgası, kalite ve model sürümüyle birlikte yaklaşık 115,7 bpm'lik ölçeklenmiş bir hız görür. Eşlenmemiş etiketler sessizce kaybolmak yerine özel bir topic'e düşer; birinin bir PLC'ye eklediği yeni etiketleri bu şekilde keşfedersiniz.

Endüstriyel DataOps'u Yaygınlaştırmak: Kontrol Listesi

1. En çok sorun yaratan tüketiciyle başlayın. Şu anda en fazla manuel temizlik gerektiren entegrasyonu seçin.
2. Ham ve modellenmiş namespace'leri ilk günden ayırın ve tüketicilere yalnızca modellenmiş olanı kullanmalarını söyleyin.
3. Eşlemeleri sürüm kontrolüne koyun ve kod gibi, incelemelerle dağıtın.
4. Yalnızca değerleri değil, kaliteyi de yayınlayın. Bayat veya aralık dışı veriyle ne yapacaklarına tüketiciler karar versin.
5. DataOps katmanının kendisini izleyin. Eşlenmemiş etiketleri, mesaj hızlarını ve bayat değer sayılarını takip edin; bunları da topic olarak yayınlayın.
6. Sahipliği açıkça belirleyin; her varlık tipi ve her saha için.
7. Tüketicileri belgeleyin; böylece bir model değişikliği ona kimin bağımlı olduğuna göre kontrol edilebilir.

Yaygın Hatalar

• DataOps'u tek seferlik bir geçiş olarak görmek. Tesisler sürekli değişir. Eşlemeyi kimse sürdürmezse birkaç ay içinde bozulur.
• Veriyi her panonun içinde temizlemek. Görselleştirme araçlarında gizlenmiş mantık yeniden kullanılamaz ve test edilemez.
• Hatalı veriyi atmak. Değerleri silmek kanıtı yok eder. Bunun yerine işaretleyin.
• Zamanı göz ardı etmek. Kaynak zaman damgaları ve saat senkronizasyonu olmadan makineler arasında sinyalleri birleştirmek tahminden ibarettir. Endüstriyel zaman serisi verisi yazısı bunun depolama ve analitik için neden önemli olduğunu açıklıyor.

Tek Bir Tesisin Ötesine Ölçeklemek

Birden fazla saha aynı yapıya yayın yaptığında DataOps katmanı aralarındaki sözleşme haline gelir. Bir bulut MQTT broker, her sahanın servislerinin modellenmiş veriyi tek bir yere yayınlamasını sağlarken plan bazlı kotalar ve çok kiracılı (multi-tenant) izolasyon ortamları birbirinden ayrı tutar. Daha büyük mimari tablo için MQTT ile endüstriyel IoT rehberi gateway'lerin, broker'ların ve tüketicilerin nasıl bir araya geldiğini gösteriyor; MQTT historian yazısı ise depolama tarafını ele alıyor.

DataOps Katmanınızı ConnectMQ Üzerinde Çalıştırın

app.connectmq.com adresinde kredi kartı gerektirmeden ücretsiz bir Serverless broker oluşturun. Edge gateway'inizi ham namespace'e yönlendirin, bağlamlandırma servisinizi YOUR_USERNAME, YOUR_PASSWORD ve YOUR_CLIENT_ID ile bir istemci olarak çalıştırın ve modellenmiş namespace'in modelinizin söylediği gibi görünüp görünmediğini doğrulamak için ConnectMQ canlı topic gezginini ve ISA-95 cihazlar görünümünü kullanın.

Sık Sorulan Sorular

Endüstriyel DataOps nedir?

Endüstriyel DataOps, OT verisini tek bir yönetilen katmanda bağlamlandırma, normalize etme, doğrulama ve yönetme pratiğidir; böylece historian'lar, MES, panolar ve analitik aynı temiz, iyi tanımlanmış bilgiyi alır.

Endüstriyel DataOps bir ürün mü yoksa bir pratik mi?

Ürünlerle desteklenebilen bir pratiktir. Eşlemeler sürümlendiği, sahiplendirildiği ve veri tüketicilere ulaşmadan önce uygulandığı sürece onu ticari DataOps platformlarıyla, Node-RED akışlarıyla veya kendi servislerinizle hayata geçirebilirsiniz.

Dönüşümleri MQTT broker mı gerçekleştirir?

Hayır. Broker mesajları dağıtır; dönüşüm, ham topic'lere abone olup modellenmiş topic'ler yayınlayan MQTT istemcilerinde çalışır. ConnectMQ ek olarak Sparkplug B'yi çözer ve onu ISA-95 topic'lerine yeniden yayınlayabilir.

Endüstriyel DataOps'un unified namespace ile ilişkisi nedir?

Unified namespace, modellenmiş verinin yayınlandığı ve keşfedildiği yerdir; DataOps ise içindeki veriyi tutarlı ve güvenilir kılan iştir. DataOps olmadan bir namespace, dağınık verilerden oluşan iyi düzenlenmiş bir koleksiyona dönüşme eğilimindedir.