Telegraf json_v2 Parser ve MQTT Topic Ayrıştırma Rehberi

Telegraf json_v2 parser, bir MQTT payload'unun tam olarak hangi bölümlerinin measurement, tag ve field olacağına karar vermenizi sağlar; topic ayrıştırma ise tanımlayıcıları doğrudan topic'in kendisinden çıkarır. Bu eğitimde ikisini de Telegraf'ın mqtt_consumer girdisi üzerinde, gerçekçi ve iç içe bir payload'dan başlayarak yapılandıracak ve çıktıyı InfluxDB'ye herhangi bir şey ulaşmadan önce satır satır doğrulayacaksınız.

Eski JSON Parser Neden Yetmiyor

Telegraf'ın orijinal json parser'ı bulduğu her şeyi field'lara düzleştirir, iç içe anahtarları alt çizgiyle birleştirir, listelemediğiniz sürece string'leri yok sayar ve tipler üzerinde sınırlı kontrol sunar. {"temperature": 21.5} gibi düz payload'lar için bu kabul edilebilir. Endüstriyel ve IoT payload'ları ise nadiren bu kadar basittir. Bir makine gateway'i tipik olarak şuna benzer bir şey yayınlar:

{
  "ts": 1760600000123,
  "device": { "id": "cnc1", "fw": "2.4.1", "serial": "SN-88231" },
  "status": "running",
  "metrics": { "spindle_rpm": 11800, "temp_c": 41.7, "vibration_mm_s": 2.3 },
  "alarms": []
}

Firmware'in tag, durumun string field, üç metriğin doğru tiplenmiş field'lar olmasını, seri numarasının yok sayılmasını ve cihaz zaman damgasının nokta zamanı olarak kullanılmasını istersiniz. json_v2 parser bu kararların her birini GJSON path sözdizimiyle açıkça ifade eder; böylece hiçbir şey veritabanına tesadüfen girmez.

Örnek Kurulum

Mesajlar ISA-95 tarzı bir topic üzerinden gelir:

connectmq/demo/a1b2c3/factory1/line1/cnc1/telemetry

Bu yedi seviyedir ve public broker'ın sekiz seviyelik sınırının içindedir. Topic zaten tesis, hat ve makine bilgisini taşıdığından yapılandırma bunları payload'a güvenmek yerine topic'ten alacaktır.

a1b2c3 yerine kendi rastgele kimliğinizi kullanarak broker.connectmq.com'a bir test mesajı yayınlayın:

mosquitto_pub -h broker.connectmq.com -p 1883 -i pub-a1b2c3 \
  -t connectmq/demo/a1b2c3/factory1/line1/cnc1/telemetry \
  -m "{\"ts\":$(date +%s%3N),\"device\":{\"id\":\"cnc1\",\"fw\":\"2.4.1\",\"serial\":\"SN-88231\"},\"status\":\"running\",\"metrics\":{\"spindle_rpm\":11800,\"temp_c\":41.7,\"vibration_mm_s\":2.3},\"alarms\":[]}"

Telegraf json_v2 Parser Yapılandırması

Bunu json_v2.conf olarak kaydedin:

[agent]
  omit_hostname = true

[[inputs.mqtt_consumer]]
  servers = ["tcp://broker.connectmq.com:1883"]
  topics = ["connectmq/demo/a1b2c3/+/+/+/telemetry"]
  client_id = "telegraf-json-a1b2c3"
  qos = 0
  topic_tag = ""
  data_format = "json_v2"

  # Tesis, hat ve makine topic'ten gelir
  [[inputs.mqtt_consumer.topic_parsing]]
    topic = "connectmq/demo/a1b2c3/+/+/+/telemetry"
    tags = "_/_/_/site/line/asset/_"

  [[inputs.mqtt_consumer.json_v2]]
    measurement_name = "machine"
    timestamp_path = "ts"
    timestamp_format = "unix_ms"

    [[inputs.mqtt_consumer.json_v2.tag]]
      path = "device.fw"
      rename = "firmware"

    [[inputs.mqtt_consumer.json_v2.field]]
      path = "status"
      type = "string"

    [[inputs.mqtt_consumer.json_v2.field]]
      path = "metrics.spindle_rpm"
      rename = "spindle_rpm"
      type = "int"

    [[inputs.mqtt_consumer.json_v2.field]]
      path = "metrics.temp_c"
      rename = "temp_c"
      type = "float"

    [[inputs.mqtt_consumer.json_v2.field]]
      path = "metrics.vibration_mm_s"
      rename = "vibration_mm_s"
      type = "float"

    [[inputs.mqtt_consumer.json_v2.field]]
      path = "alarms.#"
      rename = "alarm_count"
      type = "int"

[[outputs.file]]
  files = ["stdout"]
  data_format = "influx"

Test modunda çalıştırın. mqtt_consumer bir servis girdisidir; bu yüzden ona mesaj alması için zaman tanıyın ve o beklerken test mesajını yayınlayın:

telegraf --config json_v2.conf --test --test-wait 20

Çıktı tek satırlık bir line protocol'dür:

machine,asset=cnc1,firmware=2.4.1,line=line1,site=factory1 status="running",spindle_rpm=11800i,temp_c=41.7,vibration_mm_s=2.3,alarm_count=0i 1760600000123000000

Eşleme Nasıl Çalışır

Path'ler GJSON sözdizimini kullanır

Her path, payload üzerinde değerlendirilen bir GJSON ifadesidir. Nokta iç içe nesnelerin içine iner; bu yüzden metrics.temp_c, metrics içindeki sıcaklığı okur. # değiştiricisi bir dizinin uzunluğunu döndürür; alarms.# ifadesinin diziyi saklamadan alarm sayısı üretmesinin nedeni budur. GJSON ayrıca tek bir eleman seçmeniz gerektiğinde readings.0.value gibi dizi indekslerini ve readings.#(sensor=="t1").value gibi sorguları da destekler.

Tag'ler, field'lar ve tipler

Tipleri açıkça tanımlamak klasik bir InfluxDB sorununu önler: bir kez tamsayı, daha sonra float olarak yazılan bir field tip çakışması nedeniyle reddedilir. Her field için bir tip seçin ve onu hiç değiştirmeyin.

rename olmadan iç içe bir path, path'in düzleştirilmiş haliyle bir field adına dönüşür ve bu da genellikle uzun, hantal adlar üretir. Yeniden adlandırmak şemayı sorgularda ve dashboard'larda okunabilir tutar.

Topic ayrıştırma hiyerarşiyi doldurur

topic_parsing bloğu topic'i / karakterinden böler ve tags içindeki desenle hizalar. Her _ bir seviyeyi atlar, her ad ise o konumdan bir tag oluşturur. Desen, topic ile tam olarak aynı sayıda segmente sahip olmalıdır; bloğun içindeki topic filtresi ise kuralın hangi mesajlara uygulanacağına karar verir ve + ile # joker karakterlerine izin verir.

Topic ayrıştırma measurement'ı da belirleyebilir ve field oluşturabilir. İşte aynı kuralın measurement'ı sabit measurement_name yerine son topic seviyesinden alan hali:

  [[inputs.mqtt_consumer.topic_parsing]]
    topic = "connectmq/demo/a1b2c3/+/+/+/+"
    measurement = "_/_/_/_/_/_/measurement"
    tags = "_/_/_/site/line/asset/_"

Bu kuralla .../cnc1/telemetry üzerindeki mesajlar telemetry measurement'ına, .../cnc1/energy üzerindeki mesajlar ise energy measurement'ına düşer. Topic ayrıştırma payload parser'dan sonra çalışır; bu nedenle topic'ten gelen measurement, measurement_name ile verileni geçersiz kılar. Topic'in kendisinde taşınan sayısal değerler için tiplerini belirlemek üzere fields ile birlikte bir [inputs.mqtt_consumer.topic_parsing.types] tablosu kullanın; ancak bu nadiren iyi bir topic tasarımıdır.

Dizileri Birden Fazla Metriğe Ayrıştırmak

Gateway'ler sıklıkla birkaç okumayı tek bir mesajda toplar:

{
  "gateway": "gw-07",
  "readings": [
    { "sensor": "t1", "ts": 1760600000100, "value": 21.3 },
    { "sensor": "t2", "ts": 1760600000105, "value": 22.0 },
    { "sensor": "h1", "ts": 1760600000110, "value": 48.5 }
  ]
}

Bir object bloğu, dizinin her elemanını kendi metriğine dönüştürür. Bunu toplu mesaj topic'ini dinleyen ayrı bir mqtt_consumer girdisine koyun; böylece readings dizisi olmayan payload'lar üzerinde hiç çalışmaz:

[[inputs.mqtt_consumer]]
  servers = ["tcp://broker.connectmq.com:1883"]
  topics = ["connectmq/demo/a1b2c3/+/+/+/batch"]
  client_id = "telegraf-batch-a1b2c3"
  topic_tag = ""
  data_format = "json_v2"

  [[inputs.mqtt_consumer.topic_parsing]]
    topic = "connectmq/demo/a1b2c3/+/+/+/batch"
    tags = "_/_/_/site/line/gateway/_"

  [[inputs.mqtt_consumer.json_v2]]
    measurement_name = "reading"

    [[inputs.mqtt_consumer.json_v2.object]]
      path = "readings"
      tags = ["sensor"]
      timestamp_key = "ts"
      timestamp_format = "unix_ms"
      [inputs.mqtt_consumer.json_v2.object.fields]
        value = "float"

Her eleman reading,gateway=gw-07,line=line1,sensor=t1,site=factory1 value=21.3 1760600000100000000 gibi bir satır üretir. tags listesi anahtarları tag'e yükseltir, timestamp_key eleman başına zamanı okur ve fields tablosu tipleri sabitler. Gateway tag'inin topic'ten geldiğine dikkat edin; bu, object bloğunun yalnızca diziye odaklanmasını sağlar.

Sorun Giderme Kontrol Listesi

1. Test modunda çıktı yok. Topic'i mosquitto_sub -h broker.connectmq.com -p 1883 -i dbg-a1b2c3 -t "connectmq/demo/a1b2c3/#" -v ile doğrulayın ve --test-wait süresi içinde yayın yaptığınızdan emin olun.
2. Çıktıda bir field eksik. GJSON path'i gerçek payload ile karşılaştırın; path'teki bir yazım hatası sessizce hiçbir şey döndürmez. Yeni Telegraf sürümleri, her zaman bulunmayan field'larda optional = true ayarını kabul eder.
3. Topic ayrıştırma segmentleriyle ilgili hata. topic filtresindeki ve tags desenindeki seviyeleri sayın; eşleşmeleri gerekir.
4. Zaman damgaları 1970'te. Payload saniye kullanıyor ama yapılandırma unix_ms diyor ya da tersi.
5. InfluxDB'de tip çakışması hataları. Önceki bir yapılandırma aynı field'ı başka bir tiple yazmış. Tipleri değiştirirken yeni bir field adı ya da yeni bir bucket kullanın.
6. Her değer string. Bir field'ın type değeri string olarak ayarlanmış ya da cihaz sayıları "41.7" gibi tırnak içinde gönderiyor. type = "float" ayarı tırnaklı sayıları dönüştürür.
7. Çok fazla seri. Seri numarası veya mesaj kimliği gibi yüksek kardinaliteli bir değer tag'e eşlenmiş. Onu field yapın ya da atın.

Üretimde Çalıştırmak

Eşleme doğru olduğunda outputs.file bloğunu kaldırın, bir [[outputs.influxdb_v2]] çıktısı ekleyin ve girdiyi özel bir broker'a geçirin. ConnectMQ gibi tamamen yönetilen bir MQTT hizmetiyle bu; servers = ["ssl://broker.connectmq.com:8883"], username, password ve portalda verilen client ID anlamına gelir. Yeni bir cihaz tipi için ayrıştırma kuralları yazmadan önce portaldaki canlı topic gezgini, gerçek bir payload'u kopyalamanın pratik bir yoludur. ConnectMQ ayrıca Sparkplug B'yi çözümler ve ISA-95 topic'lerine yeniden yayınlayabilir; böylece Telegraf temiz ISA-95 topic'lerine abone olabilir. Path yazmadan önce yeniden yayınlanan bir payload'u portalda inceleyin.

Girdi eklentisinin temelleri için Telegraf MQTT consumer eğitimiyle başlayın. Grafana dahil tüm veri hattını çalıştırmak için TIG stack Docker Compose kurulumunu kullanın; cihaz tarafı kütüphaneleri seçmek için MQTT istemci kütüphaneleri rehberine bakın. Resmi json_v2 parser referansı tüm seçenekleri listeler.

ConnectMQ ile Başlayın

Ayrıştırma kurallarınızı örnek payload'larla ücretsiz public MQTT broker üzerinde test edin, ardından gerçek makineler için özel kimlik bilgilerine geçin. ConnectMQ, TLS'li ve canlı topic ile payload'ları gösteren bir portala sahip yönetilen bir MQTT 5 broker çalıştırır. app.connectmq.com adresinde kredi kartı gerektirmeden ücretsiz bir Serverless broker oluşturun. Barındırma seçeneklerine daha geniş bir bakış için bulut MQTT broker rehberine göz atın.

Sık Sorulan Sorular

Telegraf'taki json ve json_v2 parser'ları arasındaki fark nedir?

Eski json parser, payload'ları sınırlı kontrolle otomatik olarak düzleştirir. json_v2 parser ise tag'ler, field'lar, tipler ve zaman damgaları için açık GJSON path'leri kullanır; bu da iç içe IoT payload'larına çok daha iyi uyar.

Telegraf tag'leri MQTT topic'inden alabilir mi?

Evet. Bir topic_parsing bloğu topic seviyelerini tag'lere, field'lara ya da measurement adına eşler. Atlamak istediğiniz seviyeler için _ kullanın.

Bir JSON dizisini ayrı metriklere nasıl ayrıştırırım?

Dizinin path'iyle bir object bloğu kullanın. Her eleman kendi metriği olur; anahtarları tag'e yükseltebilir ve eleman başına bir zaman damgası belirleyebilirsiniz.

Düz payload'lar için json_v2'ye ihtiyacım var mı?

Hayır. Düz sayısal payload'lar için eski parser gayet iyi çalışır. String'lere, açık tiplere, yeniden adlandırılmış field'lara ya da iç içe yapılara ihtiyaç duyduğunuzda json_v2'ye geçin.