Python ile MQTT Verisini InfluxDB'ye Kaydetme

MQTT verisini InfluxDB'ye kaydetmek için her mesajı doğru measurement, tag, field ve zaman damgasıyla bir zaman serisi noktasına dönüştüren bir abone gerekir. Bu eğitimde o aboneyi Python'da paho-mqtt ve resmi InfluxDB istemcisiyle yazacak, tag'leri topic seviyelerinden otomatik türetecek, toplu yazma yapacak ve gerçek cihazların ürettiği payload sorunlarını ele alacaksınız.

Neden Bir Python Köprüsü Yazmalı

Telegraf yalnızca yapılandırmayla MQTT'ye abone olup InfluxDB'ye yazabilir ve birçok ekip için doğru tercih budur. Kısa bir Python köprüsü ise yapılandırmanın rahatça ifade edemediği mantığa ihtiyaç duyduğunuzda değerlidir: değerleri fiziksel sınırlara göre doğrulamak, birim dönüştürmek, yinelenenleri atmak, noktaları bir eşleme tablosundan gelen meta veriyle zenginleştirmek ya da aynı topic'lerde birden fazla payload biçimini işlemek. Ayrıca okunması, test edilmesi ve genişletilmesi kolaydır.

Köprü sıradan bir MQTT istemcisidir. Broker'a bağlanır, abone olur ve InfluxDB'ye HTTP API'si üzerinden yazar. Broker tarafında herhangi bir eklenti ya da entegrasyon yoktur; bu da aynı kodun broker.connectmq.com adresindeki public broker ve özel bir ConnectMQ broker dahil, standartlara uygun her broker ile çalıştığı anlamına gelir.

Veri Modeli: Topic'lerden Tag'lere

InfluxDB veriyi measurement, tag ve field olarak düzenler. Tag'ler filtreleme ve gruplama yaptığınız, indekslenmiş metinlerdir. Field'lar ise grafiğe döktüğünüz ve topladığınız değerlerdir. İyi tasarlanmış bir MQTT topic'i ihtiyacınız olan tag'leri zaten içerir:

connectmq/demo/a1b2c3/line1/sensor-0042/telemetry
   |       |     |      |        |          |
 ignore ignore ignore  line    device   measurement

Field'ları ve zaman damgasını payload sağlar:

{"device": "sensor-0042", "ts": 1760600000123, "temperature": 21.84, "humidity": 44.2}

Ortaya çıkan InfluxDB noktası line protocol'de şöyle görünür:

telemetry,line=line1,device=sensor-0042 temperature=21.84,humidity=44.2 1760600000123000000

Topic seviyelerini konuma göre eşlemek köprüyü genel tutar: yeni bir hat ya da cihaz eklediğinizde kodda hiçbir şey değişmez.

InfluxDB'yi Kurun

Elinizde InfluxDB 2 yoksa Docker ile çalıştırın. Init değişkenleri ilk başlatmada bir organizasyon, bir bucket ve bir yönetici token'ı oluşturur.

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=iot \
  -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

Yerel bir testin ötesindeki her durumda InfluxDB arayüzünde yalnızca sensors bucket'ına yazma yetkisi olan ayrı bir token oluşturun ve yönetici token'ı yerine onu kullanın.

Python paketlerini kurun:

pip install "paho-mqtt>=2.0" influxdb-client

MQTT Verisini InfluxDB'ye Kaydeden Eksiksiz Köprü

Aşağıdakini mqtt_influx_bridge.py olarak kaydedin. Yapılandırma ortam değişkenlerinden gelir; böylece aynı dosya hem testte hem üretimde çalışır.

import json
import logging
import os
import signal
import time
import uuid

import paho.mqtt.client as mqtt
from influxdb_client import InfluxDBClient, Point, WritePrecision, WriteOptions

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("bridge")

# --- yapılandırma ----------------------------------------------------------
MQTT_HOST = os.getenv("MQTT_HOST", "broker.connectmq.com")
MQTT_PORT = int(os.getenv("MQTT_PORT", "1883"))
MQTT_TLS = os.getenv("MQTT_TLS", "0") == "1"
MQTT_USER = os.getenv("MQTT_USERNAME")
MQTT_PASS = os.getenv("MQTT_PASSWORD")
MQTT_CLIENT_ID = os.getenv("MQTT_CLIENT_ID", f"influx-bridge-{uuid.uuid4().hex[:8]}")
MQTT_TOPIC = os.getenv("MQTT_TOPIC", "connectmq/demo/a1b2c3/+/+/telemetry")
# Her topic seviyesi için bir ad: "_" seviyeyi yok sayar, "@measurement" measurement adını verir
TOPIC_SCHEMA = os.getenv("TOPIC_SCHEMA", "_/_/_/line/device/@measurement").split("/")

INFLUX_URL = os.getenv("INFLUX_URL", "http://localhost:8086")
INFLUX_TOKEN = os.environ["INFLUX_TOKEN"]
INFLUX_ORG = os.getenv("INFLUX_ORG", "iot")
INFLUX_BUCKET = os.getenv("INFLUX_BUCKET", "sensors")

SKIP_KEYS = {"ts", "device"}  # field olmayan payload anahtarları

# --- InfluxDB --------------------------------------------------------------
influx = InfluxDBClient(url=INFLUX_URL, token=INFLUX_TOKEN, org=INFLUX_ORG)
write_api = influx.write_api(write_options=WriteOptions(
    batch_size=500, flush_interval=5_000, retry_interval=5_000, max_retries=5))

stats = {"written": 0, "skipped": 0}


def to_point(topic: str, payload: bytes):
    levels = topic.split("/")
    if len(levels) != len(TOPIC_SCHEMA):
        return None
    try:
        data = json.loads(payload)
    except (UnicodeDecodeError, json.JSONDecodeError):
        return None
    if not isinstance(data, dict):
        return None

    measurement = "mqtt"
    tags = {}
    for name, value in zip(TOPIC_SCHEMA, levels):
        if name == "_":
            continue
        if name == "@measurement":
            measurement = value
        else:
            tags[name] = value

    point = Point(measurement)
    for k, v in tags.items():
        point = point.tag(k, v)

    fields = 0
    for key, value in data.items():
        if key in SKIP_KEYS:
            continue
        if isinstance(value, bool):
            point = point.field(key, value)
        elif isinstance(value, (int, float)):
            point = point.field(key, float(value))  # int/float tip çakışmalarını önler
        elif isinstance(value, str) and len(value) <= 64:
            point = point.field(key, value)
        else:
            continue
        fields += 1
    if fields == 0:
        return None

    ts = data.get("ts")
    if isinstance(ts, (int, float)) and ts > 1e12:  # milisaniyeye benziyor
        point = point.time(int(ts), WritePrecision.MS)
    else:
        point = point.time(time.time_ns(), WritePrecision.NS)
    return point


def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        log.error("MQTT connect failed: %s", reason_code)
        return
    client.subscribe(MQTT_TOPIC, qos=1)
    log.info("subscribed to %s", MQTT_TOPIC)


def on_message(client, userdata, msg):
    point = to_point(msg.topic, msg.payload)
    if point is None:
        stats["skipped"] += 1
        return
    write_api.write(bucket=INFLUX_BUCKET, record=point)  # tamponlu, bloklamaz
    stats["written"] += 1


def main():
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=MQTT_CLIENT_ID)
    if MQTT_USER:
        client.username_pw_set(MQTT_USER, MQTT_PASS)
    if MQTT_TLS:
        client.tls_set()
    client.on_connect = on_connect
    client.on_message = on_message
    client.reconnect_delay_set(min_delay=1, max_delay=60)
    client.connect_async(MQTT_HOST, MQTT_PORT, keepalive=30)
    client.loop_start()

    running = True

    def stop(*_):
        nonlocal running
        running = False

    signal.signal(signal.SIGINT, stop)
    signal.signal(signal.SIGTERM, stop)

    while running:
        time.sleep(30)
        log.info("written=%d skipped=%d", stats["written"], stats["skipped"])

    client.loop_stop()
    client.disconnect()
    write_api.close()  # bekleyen noktaları yazar
    influx.close()


if __name__ == "__main__":
    main()

Köprüyü çalıştırın ve bir veri kaynağı başlatın. MQTT cihaz simülatörü tam olarak bu biçimde yayın yapar; MQTT_TOPIC değerini onun önekine ayarlayın:

export INFLUX_TOKEN=my-super-secret-token
export MQTT_TOPIC="connectmq/demo/3f9a1c2e/+/+/telemetry"
python mqtt_influx_bridge.py

Ardından veriyi InfluxDB Data Explorer'da ya da CLI ile kontrol edin:

influx query 'from(bucket: "sensors")
  |> range(start: -10m)
  |> filter(fn: (r) => r._measurement == "telemetry" and r._field == "temperature")
  |> group(columns: ["line"])
  |> mean()' --org iot --token my-super-secret-token

Tasarım Kararlarının Açıklaması

Toplu yazma ağ iş parçacığını hızlı tutar

WriteOptions ile yapılandırılmış write_api, noktaları tamponlar ve 500 nokta biriktiğinde ya da her beş saniyede bir arka planda yazar. on_message callback'i hiçbir zaman bir HTTP isteğini beklemez; böylece paho keep-alive'ları ve gelen mesajları işlemeye devam edebilir. Mesaj başına senkron yazma birkaç sensör için işe yarar, ancak mesaj hızı arttıkça hızla geride kalır. Kapanışta write_api.close() çağrısı tamponda kalan her şeyi yazar.

Sayıları float'a çevirmek tip çakışmalarını önler

InfluxDB, bir field'ın tipini bir shard'a ilk kez yazıldığında sabitler. Bir cihaz bir kez "temperature": 21, daha sonra 21.5 gönderirse ikinci yazma tip çakışması nedeniyle reddedilir. Her sayıyı float'a çevirmek bu hata sınıfını tamamen ortadan kaldırır.

Tag'lerin kardinalitesi düşük kalmalı

Tag'lere yalnızca kararlı tanımlayıcılar girmelidir: tesis, hat, cihaz. Zaman damgalarını, sıra numaralarını veya serbest metin mesajlarını asla tag'e koymayın; çünkü her benzersiz kombinasyon yeni bir seri oluşturur ve seri sayısı bellek kullanımını ile sorgu hızını belirler.

Sabit kodlanmış konumlar yerine topic şeması

TOPIC_SCHEMA hiyerarşiyi bir kez tanımlar; böylece connectmq/factory1/production/line1/cnc1/telemetry gibi bir ISA-95 yolu benzeri farklı bir topic düzeni için yalnızca yeni bir ortam değişkeni gerekir: TOPIC_SCHEMA="_/site/area/line/device/@measurement".

Diğer Payload Biçimlerini İşlemek

Gerçek cihaz filoları nadiren tek bir biçimde uzlaşır. İki varyasyon sürekli karşınıza çıkar ve ikisi de birkaç satırla to_point içine sığar.

Yalın sayılar. Birçok basit cihaz topic başına tek bir değer yayınlar; örneğin .../room1/temperature üzerinde 21.84. Burada son topic seviyesi measurement değil, field adıdır. JSON ayrıştırmadan önce float(payload) deneyin; başarılı olursa sensor gibi sabit bir measurement ile bir nokta oluşturun ve son topic seviyesini field anahtarı olarak kullanın.

İç içe nesneler. Endüstriyel gateway'ler sıklıkla {"metrics": {"spindle_rpm": 11800, "temp_c": 41.7}} gibi yapılar gönderir. Bunları field döngüsünden önce düzleştirin ve anahtarları alt çizgiyle birleştirin; böylece field metrics_spindle_rpm olur. Düzleştirme derinliğini sınırlı tutun; derin iç içe payload'lar genellikle zaman serisi veritabanına hiç ait olmayan yapılandırma ya da tanılama verileri içerir.

Ne eklerseniz ekleyin şu kuralı koruyun: bilinmeyen veya hatalı mesajlar sayılır ve atlanır, callback içinde asla istisna fırlatmalarına izin verilmez. Tek bir hatalı cihaz, köprüyü herkes için durduramamalıdır.

Sorun Giderme

Çevrimdışı tamponlama dahil daha kapsamlı yeniden bağlanma yönetimi için sağlam paho MQTT yeniden bağlanma mantığı yazısına bakın. Bunun yerine yalnızca yapılandırmayla ilerlemek isterseniz Telegraf MQTT consumer rehberi aynı veri modelini kod yazmadan oluşturur.

Üretim Kontrol Listesi

1. MQTT_TLS=1 ve broker portalınızdaki kimlik bilgileriyle 8883 portu üzerinden TLS ile bağlanın.
2. Tek bir bucket ile sınırlandırılmış özel bir InfluxDB token'ı kullanın.
3. Bucket'a, ham veriye gerçekte ne kadar süre ihtiyaç duyduğunuzla örtüşen bir saklama süresi tanımlayın.
4. Köprüyü systemd altında ya da otomatik yeniden başlatmalı bir container olarak çalıştırın.
5. Yeniden başlatmanın okumaları sessizce düşürmemesi için QoS 1 ve sabit bir client ID ile abone olun.
6. written ve skipped sayaçlarını izleyin; atlanan mesajlardaki ani artış genellikle bir firmware değişikliğine işaret eder.

ConnectMQ gibi tamamen yönetilen bir MQTT hizmeti ile birinci ve beşinci maddeler yalnızca yapılandırmadan ibarettir: YOUR_USERNAME, YOUR_PASSWORD ve YOUR_CLIENT_ID değerlerini portaldan ortam değişkenlerine kopyalayın. Köprünün kendisi değişmez. InfluxDB 3'ün farklı bir Python istemcisi kullandığını unutmayın; bu rehberdeki veri modeli yine geçerlidir, ancak yazma çağrıları farklıdır. Burada kullanılan 2.x istemcisi influxdb-client dokümantasyonunda anlatılır. Cihaz tarafındaki alternatifler için MQTT istemci kütüphaneleri karşılaştırmasına göz atın.

ConnectMQ ile Başlayın

Köprüyü önce simüle edilmiş veriyle ücretsiz public MQTT broker üzerinde deneyin. Gerçek cihazlar için ConnectMQ, TLS, MQTT 5 ve köprünüzün hangi topic'leri alacağını tam olarak gösteren canlı bir topic gezgini içeren özel bir broker sunar. Kredi kartı gerektirmeyen ücretsiz bir Serverless broker için app.connectmq.com adresinden kaydolun ve büyümeyi planlıyorsanız buluttaki MQTT broker modelini okuyun.

Sık Sorulan Sorular

MQTT verisini InfluxDB'ye kaydetmek için Telegraf mı Python mu kullanmalıyım?

Payload'larınız yapılandırmayla karşılanabiliyorsa Telegraf'ı kullanın, çünkü kod gerektirmez. Doğrulama, zenginleştirme, birim dönüştürme ya da birden fazla payload biçimi gerektiğinde Python'u kullanın.

Hangi MQTT topic seviyeleri tag olmalı?

Tesis, hat ve cihaz gibi kararlı tanımlayıcılar. Mesaj başına benzersiz olan her şeyden kaçının, çünkü yüksek tag kardinalitesi InfluxDB'yi yavaşlatır.

Neden bazı noktalarım reddediliyor?

En yaygın neden, bir field'ın önce tamsayı, sonra float ya da string olarak yazıldığı field tipi çakışmasıdır. Sayıları tutarlı biçimde float'a çevirin.

Köprü JSON olmayan payload'ları işleyebilir mi?

Bu haliyle hayır; onları atlar ve sayar. Cihazlarınız düz sayılar ya da başka biçimler gönderiyorsa bunları ayrıştırmak için to_point fonksiyonunu genişletin.