Parquet'tan Gömülü Cihazlara Config Kayıtlarını İdempotent Şekilde Senkronize Etme

Filo cihazları genellikle bağlı düğüm/sensörlerin yerel bir kaydını tutar (seri numarası, bus adresi, isim, bölge, kalibrasyon değerleri). Yetkili veri bir veritabanında yaşadığında temiz aktarım şudur: Parquet’e dışa aktar → her cihazın HTTP API’sine karşı idempotent bir senkronizasyon script’i çalıştır. Bu yazı, ikinci adım için yeniden kullanılabilir bir desen ve pratikte her zaman yaşanan kısmi hataların nasıl ele alınacağını gösteriyor.

Tasarım kuralları

İdempotent: Script’i iki kez çalıştırmak aynı duruma yakınsamalı — “create or update”, asla “create or fail” değil.
Adres-farkında oluşturma: Geçerli bus adresi olmayan kayıtlar zorla oluşturulamaz — çalışmayı iptal etmek yerine net bir raporla atlanmalı.
Alan-bazlı güncellemeler: Her özniteliği (isim, bölge, setpoint’ler) kendi endpoint’i üzerinden gönder, böylece kısmi kayıt girdileri bile tamamen dolar.
Doğrulama: Sonrasında, cihazın bildirdiği kayıt sayısını Parquet satır sayısıyla karşılaştır.

Senkronizasyon script’i

sync_devices.py

#!/usr/bin/env python3
"""Bir Parquet dosyasındaki düğüm config kayıtlarını cihaz API'sine idempotent şekilde senkronize et."""
import sys
import pandas as pd
import requests

HOST = sys.argv[1]          # örn. 10.0.0.42
PARQUET = sys.argv[2]
BASE = f"http://{HOST}/api"
TIMEOUT = 10

df = pd.read_parquet(PARQUET)
ok, skipped, failed = 0, [], []

def post(path, payload):
    r = requests.post(f"{BASE}{path}", json=payload, timeout=TIMEOUT)
    return r

for _, row in df.iterrows():
    serial = str(row["SerialNo"])
    addr = row.get("CANAddress")

    # Bus'ta adreslenemeyen kayıtları atla —
    # ancak cihaz kendisi keşfettiğinde ortaya çıkabilirler.
    if pd.isna(addr) or int(addr) == 0:
        skipped.append(serial)
        continue

    try:
        # Oluştur-veya-birleştir kimliği: seri numarası zaten var
        # olsa da adrese yeni bağlansa da endpoint 200 döner.
        r = post("/nodes/merge", {
            "serial": serial,
            "address": int(addr),
        })
        if r.status_code == 404:
            skipped.append(f"{serial} (not in registry, address {int(addr)})")
            continue
        r.raise_for_status()

        # Alan-bazlı güncellemeler — her biri idempotent.
        post("/nodes/set-type",  {"serial": serial, "type": row["Type"]}).raise_for_status()
        for field, ep in [("Name", "name"), ("Name2", "name2"),
                          ("Name3", "name3"), ("Name4", "name4"),
                          ("ZoneName", "zone")]:
            val = row.get(field)
            if pd.notna(val) and str(val):
                post(f"/nodes/set-{ep}", {"serial": serial, "value": str(val)}).raise_for_status()
        if pd.notna(row.get("SetVoltage")):
            post("/nodes/set-voltage", {"serial": serial,
                                        "value": float(row["SetVoltage"])}).raise_for_status()
        if pd.notna(row.get("SetCurrent")):
            post("/nodes/set-current", {"serial": serial,
                                        "value": float(row["SetCurrent"])}).raise_for_status()
        ok += 1
    except requests.RequestException as e:
        failed.append((serial, str(e)))

print(f"OK: {ok}/{len(df)}  skipped: {len(skipped)}  failed: {len(failed)}")
for s in skipped:
    print(f"  skip {s}")
for s, e in failed:
    print(f"  FAIL {s}: {e}")

Doğrulama

verify.py

import requests, pandas as pd
device = requests.get(f"http://{HOST}/api/nodes", timeout=10).json()
expected = len(pd.read_parquet(PARQUET))
print(f"device: {len(device)} records, parquet: {expected}")

Sayılar meşru şekilde farklı olabilir: cihaz ekstra otomatik-keşfedilmiş düğümler tutabilir ve atlanan kayıtlar (geçerli adres yok) ancak bus onları bulduğunda ortaya çıkar.

Kısmi hataları ele alma

Belirti	Anlam	Eylem
Merge’de `404`/`ESP_ERR_NOT_FOUND`	Seri numarası henüz kayıtta değil VE adres eksik/0	Bus keşfini bekle, yeniden çalıştır — geri kalanı zaten deploy edildi
Bir kayıtta HTTP 400	Bozuk alan değeri	Satırı düzelt, yeniden çalıştır (idempotent)
Cihaza ulaşılamıyor	Ağ/VPN sorunu	Yine de Parquet’i filo dizinine deploy et; ulaşılabilir olunca senkronize et

Önemli özellik: keşiften sonra script’i yeniden çalıştırmak, cihazın kendi bulduğu düğümlerde isimleri/bölgeleri/setpoint’leri doldurur — hiçbir şey kaybolmaz, hiçbir şey çoğaltılmaz.

İlgili

How to Convert a MariaDB SQL Dump to Parquet Using a Docker Container — bu script’in tükettiği Parquet’i üretme.