Verwenden Sie WebAssembly (WASM)-Transformationen in Datenflussdiagrammen

Azure IoT Einsatz Datenflussdiagramme enthalten integrierte Transformationen für allgemeine Verarbeitungsaufgaben wie Zuordnung, Filterung und Aggregation. Wenn Sie benutzerdefinierte Logik benötigen, die über die integrierten Transformationen hinausgeht, können Sie WebAssembly -Module (WASM) als benutzerdefinierte Transformationen in Ihren Datenflussdiagrammpipelines bereitstellen.

Voraussetzungen

• Eine Instanz von Azure IoT Einsatz, die in einem Kubernetes-Cluster bereitgestellt wird. Weitere Informationen finden Sie unter Deploy Azure IoT Einsatz.

• Konfigurieren Sie einen Registrierungsendpunkt für den Zugriff auf WASM-Module und Graphdefinitionen. Sie haben zwei Optionen:
  • Schnellstart mit der öffentlichen Registry: Erstellen Sie einen Registry-Endpunkt für ghcr.io mit anonymer Authentifizierung. Anweisungen finden Sie unter Verwenden vordefinierter Module aus einer öffentlichen Registrierung.
  • Private Registrierung: Richten Sie Ihre eigene Containerregistrierung ein, und pushen Sie die Beispielmodule, indem Sie die Anleitungen in "Deploy WebAssembly(WASM)"-Module und -Diagrammdefinitionen ausführen.

Die Azure CLI-Beispiele in diesem Artikel verwenden Umgebungsvariablen, sodass Sie jeden Wert nur einmal festlegen und die Befehle dann unverändert kopieren und einfügen können. Wenn Sie die Azure IoT Einsatz Codespaces-Umgebung aus dem Quickstart verwenden, sind diese Variablen bereits für Sie festgelegt und Sie können diesen Schritt überspringen. Ansonsten setze die folgenden Umgebungsvariablen in deiner Shell, bevor du die Befehle ausführst.

Die folgenden Skripte legen die am häufigsten verwendeten Umweltvariablen fest:

SUBSCRIPTION_ID=<subscription-id>
RESOURCE_GROUP=<resource-group-name>
AIO_INSTANCE_NAME=<instance-name>
CLUSTER_NAME=<cluster-name>
LOCATION=<region>

$SUBSCRIPTION_ID = "<subscription-id>"
$RESOURCE_GROUP = "<resource-group-name>"
$AIO_INSTANCE_NAME = "<instance-name>"
$CLUSTER_NAME = "<cluster-name>"
$LOCATION = "<region>"

Du musst nur die Variablen festlegen, die dieser Artikel verwendet. Dieser Artikel könnte zusätzliche Umgebungsvariablen für die von Ihnen gewählten Ressourcennamen verwenden. Der Artikel erklärt, wie man sie dort platziert, wo sie eingeführt werden.

Dieser Artikel verwendet außerdem die folgenden Umgebungsvariablen für die von Ihnen gewählten Werte: REGISTRY_ENDPOINT (Name des Registry-Endpunkts), GRAPH_NAME (Name des Datenflussgraphen), PROFILE (Name des Datenflussprofils) und REGISTRY_HOST (Container-Registry-Hostname). Stelle jede einzelne ein, bevor du die zugehörigen Befehle ausführst.

Überblick

Mithilfe von WebAssembly(WASM)-Modulen in Azure IoT Einsatz Datenflussdiagrammen können Sie Daten am Edge mit hoher Leistung und Sicherheit verarbeiten. WASM wird in einer Sandkastenumgebung ausgeführt und unterstützt Rust und Python.

Ein Datenfluss ist eine Pipeline, die Daten mithilfe integrierter Transformationen zwischen Endpunkten bewegt und transformiert. Ein Datenflussdiagramm erweitert Datenflüsse mit komponierbaren Verarbeitungsschritten. Azure IoT Einsatz bietet built-in-Datenflussdiagramme für allgemeine Vorgänge wie Zuordnung, Filterung, Verzweigung und Aggregation. Für benutzerdefinierte Verarbeitungslogik implementieren Sie WebAssembly-Module, wie in diesem Artikel beschrieben. Datenflussdiagramme verwenden YAML-Diagrammdefinitionen, die angeben, wie Operatoren eine Verbindung herstellen. Die Datenflussdiagrammressource umschließt diese Definition und ordnet seine abstrakten Quell- und Senkenvorgänge konkreten Endpunkten wie MQTT-Themen und Kafka-Themen zu.

Funktionsweise von WASM-Datenflussdiagrammen

Die WASM-Datenflussimplementierung folgt diesem Workflow:

1. Entwickeln Sie WASM-Module: Schreiben Sie benutzerdefinierte Verarbeitungslogik in einer unterstützten Sprache, und kompilieren Sie sie in das WebAssembly Component Model-Format. Weitere Informationen finden Sie unter WASM-Module für Datenflüsse erstellen.
2. Entwickeln Sie graphische Definition: Definieren Sie, wie Daten mithilfe von YAML-Konfigurationsdateien durch die Module verschoben werden. Weitere Informationen finden Sie unter Konfigurieren von WebAssembly-Diagrammdefinitionen.
3. Speichern Sie Artefakte in der Registrierung: Übertragen Sie die kompilierten WASM-Module und Graphdefinitionen mithilfe von OCI-kompatiblen Tools wie ORAS in eine Containerregistrierung. Weitere Informationen finden Sie unter Deploy WebAssembly (WASM)-Module und -Diagrammdefinitionen.
4. Registrierungsendpunkte konfigurieren: Authentifizierungs- und Verbindungsdetails einrichten, damit Azure IoT Einsatz auf die Containerregistrierung zugreifen kann. Weitere Informationen finden Sie unter Konfigurieren von Registrierungsendpunkten.
5. Datenflussgraph erstellen: Verwenden Sie die Operations Experience-Web-UI oder Bicep-Dateien, um einen Datenfluss zu definieren, der eine Graphdefinition nutzt.
6. Deploy and execute: Azure IoT Einsatz ruft Graphdefinitionen und WASM-Module aus der Containerregistrierung ab und führt sie aus.

Die folgenden Beispiele zeigen, wie SIE WASM-Datenflussdiagramme für gängige Szenarien konfigurieren. In den Beispielen werden hartcodierte Werte und vereinfachte Konfigurationen verwendet, damit Sie schnell loslegen können.

Beispiel 1: Grundlegende Bereitstellung mit einem WASM-Modul

In diesem Beispiel werden Temperaturdaten von Fahrenheit in Celsius mithilfe eines WASM-Moduls konvertiert. Der Quellcode des temperaturemoduls ist auf GitHub verfügbar. Wenn Sie die Beispielschritte in "Deploy WebAssembly(WASM)"-Module und -Diagrammdefinitionen befolgt haben, befinden sich die Diagrammdefinition und das graph-simple:1.0.0 vorkompilierte temperature:1.0.0 Modul bereits in Ihrer Containerregistrierung. Der Beispieldiagrammartefaktpfad ist azure-samples/explore-iot-operations/graph-simple:1.0.0. Verwenden Sie diesen Pfad für die öffentlichen GHCR-Beispiele oder wenn Sie die Beispielartefakte mithilfe desselben Repositorypfads in Ihre eigene Registrierung kopieren.

Funktionsweise des einfachen Diagramms

Die Graph-Definition erstellt eine einfache, dreistufige Pipeline:

1. Quelle: Empfängt Temperaturdaten von MQTT
2. Karte: Verarbeitet Daten mit dem Temperatur-WASM-Modul
3. Sink: Sendet konvertierte Daten zurück an MQTT

Weitere Informationen zur Funktionsweise der einfachen Diagrammdefinition und deren Struktur finden Sie unter Beispiel 1: Einfache Diagrammdefinition.

Eingabeformat:

{"temperature": {"value": 100.0, "unit": "F"}}

Ausgabeformat:

{"temperature": {"value": 37.8, "unit": "C"}}

Die folgende Konfiguration erstellt ein Datenflussdiagramm, das diese Temperaturkonvertierungspipeline verwendet. Das Datenflussdiagramm verweist auf die graph-simple:1.0.0 YAML-Diagrammdefinition und ruft das Temperaturmodul aus Ihrer Containerregistrierung ab. Das Azure Beispieldiagramm wird unter dem Repositorypfad azure-samples/explore-iot-operations gespeichert. Schließen Sie daher diesen Pfad in den Wert artifact ein.

Konfigurieren des Datenflussdiagramms

Diese Konfiguration definiert drei Knoten, die den Temperaturkonvertierungsworkflow implementieren: einen Quellknoten, der eingehende Temperaturdaten abonniert, einen Diagrammverarbeitungsknoten, der das WASM-Modul ausführt, und einen Zielknoten, der die konvertierten Ergebnisse veröffentlicht.

Die Datenflussdiagrammressource umschließt das Diagrammdefinitionsartefakt und verbindet seine abstrakten Quell- und Sinkvorgänge mit konkreten Endpunkten:

• Der source-Vorgang der Graphdefinition verbindet sich mit dem Quellknoten des Datenflusses (MQTT-Topic)
• Der Vorgang sink Graphdefinition stellt eine Verbindung mit dem Zielknoten des Datenflusses (MQTT-Thema) her.
• Die Verarbeitungsvorgänge der Graphdefinition werden im Diagrammverarbeitungsknoten ausgeführt.

Mit dieser Trennung können Sie dieselbe Diagrammdefinition mit verschiedenen Endpunkten in allen Umgebungen bereitstellen und gleichzeitig die Verarbeitungslogik unverändert beibehalten.

{
  "mode": "Enabled",
  "requestDiskPersistence": "Disabled",
  "nodes": [
    {
      "nodeType": "Source",
      "name": "temperature-source",
      "sourceSettings": {
        "endpointRef": "default",
        "dataSources": [
          "sensor/temperature/raw"
        ]
      }
    },
    {
      "nodeType": "Graph",
      "name": "temperature-processor",
      "graphSettings": {
        "registryEndpointRef": "<REGISTRY_ENDPOINT_NAME>",
        "artifact": "azure-samples/explore-iot-operations/graph-simple:1.0.0",
        "configuration": [
          {
            "key": "key1",
            "value": "example-value"
          }
        ]
      }
    },
    {
      "nodeType": "Destination",
      "name": "temperature-destination",
      "destinationSettings": {
        "endpointRef": "default",
        "dataDestination": "sensor/temperature/processed"
      }
    }
  ],
  "nodeConnections": [
    {
      "from": {
        "name": "temperature-source"
      },
      "to": {
        "name": "temperature-processor"
      }
    },
    {
      "from": {
        "name": "temperature-processor"
      },
      "to": {
        "name": "temperature-destination"
      }
    }
  ]
}

az iot ops dataflowgraph apply \
  --name $GRAPH_NAME \
  --instance $AIO_INSTANCE_NAME \
  --resource-group $RESOURCE_GROUP \
  --config-file graph.json

param aioInstanceName string = '<AIO_INSTANCE_NAME>'
param customLocationName string = '<CUSTOM_LOCATION_NAME>'
param dataflowGraphName string = '<GRAPH_NAME>'
param registryEndpointName string = '<REGISTRY_ENDPOINT_NAME>'

resource aioInstance 'Microsoft.IoTOperations/instances@2026-03-01' existing = {
  name: aioInstanceName
}

resource customLocation 'Microsoft.ExtendedLocation/customLocations@2021-08-31-preview' existing = {
  name: customLocationName
}

resource defaultDataflowProfile 'Microsoft.IoTOperations/instances/dataflowProfiles@2026-03-01' existing = {
  parent: aioInstance
  name: 'default'
}

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2026-03-01' = {
  parent: defaultDataflowProfile
  name: dataflowGraphName
  extendedLocation: {
    name: customLocation.id
    type: 'CustomLocation'
  }
  properties: {
    mode: 'Enabled'
    requestDiskPersistence: 'Disabled'
    nodes: [
      {
        nodeType: 'Source'
        name: 'temperature-source'
        sourceSettings: {
          endpointRef: 'default'
          dataSources: [
            'sensor/temperature/raw'
          ]
        }
      }
      {
        nodeType: 'Graph'
        name: 'temperature-processor'
        graphSettings: {
          registryEndpointRef: registryEndpointName
          artifact: 'azure-samples/explore-iot-operations/graph-simple:1.0.0'
          configuration: [
            {
              key: 'key1'
              value: 'example-value'
            }
          ]
        }
      }
      {
        nodeType: 'Destination'
        name: 'temperature-destination'
        destinationSettings: {
          endpointRef: 'default'
          dataDestination: 'sensor/temperature/processed'
        }
      }
    ]
    nodeConnections: [
      {
        from: {
          name: 'temperature-source'
        }
        to: {
          name: 'temperature-processor'
        }
      }
      {
        from: {
          name: 'temperature-processor'
        }
        to: {
          name: 'temperature-destination'
        }
      }
    ]
  }
}

apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowGraph
metadata:
  name: <GRAPH_NAME>
  namespace: azure-iot-operations
spec:
  requestDiskPersistence: Disabled
  profileRef: default

  nodes:
    - nodeType: Source
      name: temperature-source
      sourceSettings:
        endpointRef: default
        dataSources:
          - sensor/temperature/raw

    - nodeType: Graph
      name: temperature-processor
      graphSettings:
        registryEndpointRef: <REGISTRY_ENDPOINT_NAME>
        artifact: azure-samples/explore-iot-operations/graph-simple:1.0.0
        configuration:
          - key: key1
            value: example-value

    - nodeType: Destination
      name: temperature-destination
      destinationSettings:
        endpointRef: default
        dataDestination: sensor/temperature/processed

  nodeConnections:
    - from:
        name: temperature-source
      to:
        name: temperature-processor

    - from:
        name: temperature-processor
      to:
        name: temperature-destination

kubectl apply -f dataflow-graph.yaml

Testen des Datenflusses

Um den Datenfluss zu testen, senden Sie MQTT-Nachrichten aus dem Cluster. Das Temperaturmodul erwartet Nachrichten in einem bestimmten JSON-Format mit einem geschachtelten temperature-Objekt, das value (numerische) und unit (Zeichenfolgen)-Felder enthält. Beispiel: {"temperature":{"value":72,"unit":"F"}}.

Stellen Sie zunächst den MQTT-Client-Pod bereit, indem Sie die Anweisungen in Testen der Konnektivität zum MQTT-Broker mit MQTT-Clients befolgen. Der MQTT-Client stellt die Authentifizierungstoken und Zertifikate bereit, um eine Verbindung mit dem Broker herzustellen. Führen Sie zum Bereitstellen des MQTT-Clients den folgenden Befehl aus:

kubectl apply -f https://raw.githubusercontent.com/Azure-Samples/explore-iot-operations/main/samples/quickstarts/mqtt-client.yaml

Senden von Temperaturmeldungen

Erstellen und ausführen Sie in der ersten Terminalsitzung ein Skript zum Senden von Temperaturdaten in Fahrenheit:

# Connect to the MQTT client pod
kubectl exec --stdin --tty mqtt-client -n azure-iot-operations -- sh -c '
# Create and run temperature.sh from within the MQTT client pod
while true; do
  # Generate a random temperature value between 0 and 6000 Fahrenheit
  random_value=$(shuf -i 0-6000 -n 1)
  payload="{\"temperature\":{\"value\":$random_value,\"unit\":\"F\"}}"

  echo "Publishing temperature: $payload"

  # Publish to the input topic
  mosquitto_pub -h aio-broker -p 18883 \
    -m "$payload" \
    -t "sensor/temperature/raw" \
    -d \
    --cafile /var/run/certs/ca.crt \
    -D PUBLISH user-property __ts $(date +%s)000:0:df \
    -D CONNECT authentication-method 'K8S-SAT' \
    -D CONNECT authentication-data $(cat /var/run/secrets/tokens/broker-sat)

  sleep 1
done'

Das Skript veröffentlicht jede Sekunde zufällige Temperaturdaten für das Thema sensor/temperature/raw. Diese sollte wie folgt aussehen:

Publishing temperature: {"temperature":{"value":1234,"unit":"F"}}
Publishing temperature: {"temperature":{"value":5678,"unit":"F"}}

Beenden Sie das Skript nicht, um die Veröffentlichung von Temperaturdaten fortzusetzen.

Abonnieren von verarbeiteten Nachrichten

Abonnieren Sie in der zweiten Terminalsitzung (ebenfalls mit dem MQTT-Clientpod verbunden) das Ausgabethema, um die konvertierten Temperaturwerte anzuzeigen:

# Connect to the MQTT client pod
kubectl exec --stdin --tty mqtt-client -n azure-iot-operations -- sh -c '
mosquitto_sub -h aio-broker -p 18883 -t "sensor/temperature/processed" --cafile /var/run/certs/ca.crt \
-D CONNECT authentication-method "K8S-SAT" \
-D CONNECT authentication-data "$(cat /var/run/secrets/tokens/broker-sat)"'

Du siehst, dass das WASM-Modul die Temperaturdaten von Fahrenheit auf Celsius umwandelt.

{"temperature":{"value":1292.2222222222222,"count":0,"max":0.0,"min":0.0,"average":0.0,"last":0.0,"unit":"C","overtemp":false}}
{"temperature":{"value":203.33333333333334,"count":0,"max":0.0,"min":0.0,"average":0.0,"last":0.0,"unit":"C","overtemp":false}}

Beispiel 2: Bereitstellen eines komplexen Diagramms

In diesem Beispiel wird ein anspruchsvoller Datenverarbeitungsworkflow veranschaulicht, der mehrere Datentypen wie Temperatur, Feuchtigkeit und Bilddaten verarbeitet. Die complex graph definition orchestriert mehrere WASM-Module, um erweiterte Analysen und Objekterkennung durchzuführen.

Funktionsweise des komplexen Diagramms

Das komplexe Diagramm verarbeitet drei Datenströme und kombiniert sie in einer erweiterten Sensoranalyse:

• Temperaturverarbeitung: Wandelt Fahrenheit in Celsius um, filtert ungültige Messwerte und berechnet Statistiken.
• Luftfeuchtigkeitsverarbeitung: Sammelt Luftfeuchtigkeitsmessungen über Zeitintervalle
• Bildverarbeitung: Führt die Objekterkennung für Kameramomentaufnahmen aus und formatiert Ergebnisse.

Weitere Informationen zur Funktionsweise der komplexen Diagrammdefinition, ihrer Struktur und des Datenflusses über mehrere Verarbeitungsphasen finden Sie unter Beispiel 2: Komplexe Diagrammdefinition.

Das Diagramm verwendet spezielle Module aus der Sammlung von Rust-Operatoren.

Konfigurieren des komplexen Datenflussdiagramms

Diese Konfiguration implementiert den Multisensorverarbeitungsworkflow mithilfe der graph-complex:1.0.0 YAML-Diagrammdefinition. Der Beispieldiagrammartefaktpfad ist azure-samples/explore-iot-operations/graph-complex:1.0.0. Beachten Sie, dass die Bereitstellung des Datenflussdiagramms mit Beispiel 1 vergleichbar ist – beide verwenden dasselbe Muster mit drei Knoten (Quelle, Diagrammprozessor, Ziel), obwohl die Verarbeitungslogik unterschiedlich ist.

Diese Ähnlichkeit tritt auf, da die Datenflussdiagrammressource als Hostumgebung fungiert, die Diagrammdefinitionen lädt und ausführt. Die eigentliche Verarbeitungslogik befindet sich in der Diagrammdefinition (graph-simple:1.0.0 oder graph-complex:1.0.0), die die YAML-Spezifikation von Vorgängen und Verbindungen zwischen WASM-Modulen enthält. Die Datenflussdiagrammressource stellt die Laufzeitinfrastruktur bereit, um die Diagrammdefinition abzurufen, die Module zu instanziieren und Daten über den definierten Workflow weiterzuleiten.

{
  "mode": "Enabled",
  "requestDiskPersistence": "Disabled",
  "nodes": [
    {
      "nodeType": "Source",
      "name": "sensor-source",
      "sourceSettings": {
        "endpointRef": "default",
        "dataSources": [
          "sensor/temperature/raw",
          "sensor/humidity/raw",
          "sensor/images/raw"
        ]
      }
    },
    {
      "nodeType": "Graph",
      "name": "sensor-processor",
      "graphSettings": {
        "registryEndpointRef": "<REGISTRY_ENDPOINT_NAME>",
        "artifact": "azure-samples/explore-iot-operations/graph-complex:1.0.0",
        "configuration": [
          {
            "key": "snapshot_topic",
            "value": "sensor/images/raw"
          },
          {
            "key": "key1",
            "value": "example-value"
          }
        ]
      }
    },
    {
      "nodeType": "Destination",
      "name": "analytics-destination",
      "destinationSettings": {
        "endpointRef": "default",
        "dataDestination": "analytics/sensor/processed"
      }
    }
  ],
  "nodeConnections": [
    {
      "from": {
        "name": "sensor-source"
      },
      "to": {
        "name": "sensor-processor"
      }
    },
    {
      "from": {
        "name": "sensor-processor"
      },
      "to": {
        "name": "analytics-destination"
      }
    }
  ]
}

az iot ops dataflowgraph apply \
  --name $GRAPH_NAME \
  --instance $AIO_INSTANCE_NAME \
  --resource-group $RESOURCE_GROUP \
  --config-file graph.json

param aioInstanceName string = '<AIO_INSTANCE_NAME>'
param customLocationName string = '<CUSTOM_LOCATION_NAME>'
param dataflowGraphName string = '<COMPLEX_GRAPH_NAME>'
param registryEndpointName string = '<REGISTRY_ENDPOINT_NAME>'

resource aioInstance 'Microsoft.IoTOperations/instances@2026-03-01' existing = {
  name: aioInstanceName
}

resource customLocation 'Microsoft.ExtendedLocation/customLocations@2021-08-31-preview' existing = {
  name: customLocationName
}

resource defaultDataflowProfile 'Microsoft.IoTOperations/instances/dataflowProfiles@2026-03-01' existing = {
  parent: aioInstance
  name: 'default'
}

resource complexDataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2026-03-01' = {
  parent: defaultDataflowProfile
  name: dataflowGraphName
  extendedLocation: {
    name: customLocation.id
    type: 'CustomLocation'
  }
  properties: {
    mode: 'Enabled'
    requestDiskPersistence: 'Disabled'
    nodes: [
      {
        nodeType: 'Source'
        name: 'sensor-source'
        sourceSettings: {
          endpointRef: 'default'
          dataSources: [
            'sensor/temperature/raw'
            'sensor/humidity/raw'
            'sensor/images/raw'
          ]
        }
      }
      {
        nodeType: 'Graph'
        name: 'sensor-processor'
        graphSettings: {
          registryEndpointRef: registryEndpointName
          artifact: 'azure-samples/explore-iot-operations/graph-complex:1.0.0'
          configuration: [
            {
              key: 'snapshot_topic'
              value: 'sensor/images/raw'
            }
            {
              key: 'key1'
              value: 'example-value'
            }
          ]
        }
      }
      {
        nodeType: 'Destination'
        name: 'analytics-destination'
        destinationSettings: {
          endpointRef: 'default'
          dataDestination: 'analytics/sensor/processed'
        }
      }
    ]
    nodeConnections: [
      {
        from: {
          name: 'sensor-source'
        }
        to: {
          name: 'sensor-processor'
        }
      }
      {
        from: {
          name: 'sensor-processor'
        }
        to: {
          name: 'analytics-destination'
        }
      }
    ]
  }
}

apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowGraph
metadata:
  name: <COMPLEX_GRAPH_NAME>
  namespace: azure-iot-operations
spec:
  requestDiskPersistence: Disabled
  profileRef: default

  nodes:
    - nodeType: Source
      name: sensor-source
      sourceSettings:
        endpointRef: default
        dataSources:
          - sensor/temperature/raw
          - sensor/humidity/raw
          - sensor/images/raw

    - nodeType: Graph
      name: sensor-processor
      graphSettings:
        registryEndpointRef: <REGISTRY_ENDPOINT_NAME>
        artifact: azure-samples/explore-iot-operations/graph-complex:1.0.0
        configuration:
          - key: snapshot_topic
            value: sensor/images/raw
          - key: key1
            value: example-value

    - nodeType: Destination
      name: analytics-destination
      destinationSettings:
        endpointRef: default
        dataDestination: analytics/sensor/processed

  nodeConnections:
    - from:
        name: sensor-source
      to:
        name: sensor-processor

    - from:
        name: sensor-processor
      to:
        name: analytics-destination

Testen des komplexen Datenflusses

Bevor Sie eine ausgabe sehen können, richten Sie die Quelldaten ein.

Hochladen von RAW-Bilddateien auf den MQTT-Client-Pod

Die Bilddateien dienen dem Modul snapshot zum Erkennen von Objekten in den Bildern. Die Dateien befinden sich im Bilder-Ordner auf GitHub.

Klonen Sie zuerst das Repository, um Zugriff auf die Bilddateien zu erhalten:

git clone https://github.com/Azure-Samples/explore-iot-operations.git
cd explore-iot-operations

Verwenden Sie den folgenden Befehl, um RAW-Bilddateien aus dem ./samples/wasm/images Ordner in den mqtt-client Pod hochzuladen:

kubectl cp ./samples/wasm/images azure-iot-operations/mqtt-client:/tmp

Überprüfen Sie, ob die Dateien hochgeladen wurden:

kubectl exec -it mqtt-client -n azure-iot-operations -- ls /tmp/images

Die Liste der Dateien sollte im Ordner /tmp/images angezeigt werden.

beaker.raw          laptop.raw          sunny2.raw
binoculars.raw      lawnmower.raw       sunny4.raw
broom.raw           milkcan.raw         thimble.raw
camera.raw          photocopier.raw     tripod.raw
computer_mouse.raw  radiator.raw        typewriter.raw
daisy3.raw          screwdriver.raw     vacuum_cleaner.raw
digital_clock.raw   sewing_machine.raw
hammer.raw          sliding_door.raw

Veröffentlichen simulierter Temperatur- und Feuchtigkeitsdaten und Senden von Bildern

Sie können die Befehle zum Veröffentlichen von Temperatur- und Feuchtigkeitsdaten kombinieren und Bilder in ein einzelnes Skript senden. Verwenden Sie den folgenden Befehl:

# Connect to the MQTT client pod and run the script
kubectl exec --stdin --tty mqtt-client -n azure-iot-operations -- sh -c '
while true; do 
  # Generate a random temperature value between 0 and 6000
  temp_value=$(shuf -i 0-6000 -n 1)
  temp_payload="{\"temperature\":{\"value\":$temp_value,\"unit\":\"F\"}}"
  echo "Publishing temperature: $temp_payload"
  mosquitto_pub -h aio-broker -p 18883 \
    -m "$temp_payload" \
    -t "sensor/temperature/raw" \
    --cafile /var/run/certs/ca.crt \
    -D CONNECT authentication-method "K8S-SAT" \
    -D CONNECT authentication-data "$(cat /var/run/secrets/tokens/broker-sat)" \
    -D PUBLISH user-property __ts $(date +%s)000:0:df

  # Generate a random humidity value between 30 and 90
  humidity_value=$(shuf -i 30-90 -n 1)
  humidity_payload="{\"humidity\":{\"value\":$humidity_value}}"
  echo "Publishing humidity: $humidity_payload"
  mosquitto_pub -h aio-broker -p 18883 \
    -m "$humidity_payload" \
    -t "sensor/humidity/raw" \
    --cafile /var/run/certs/ca.crt \
    -D CONNECT authentication-method "K8S-SAT" \
    -D CONNECT authentication-data "$(cat /var/run/secrets/tokens/broker-sat)" \
    -D PUBLISH user-property __ts $(date +%s)000:0:df

  # Send an image every 2 seconds
  if [ $(( $(date +%s) % 2 )) -eq 0 ]; then
    file=$(ls /tmp/images/*.raw | shuf -n 1)
    echo "Sending file: $file"
    mosquitto_pub -h aio-broker -p 18883 \
      -f $file \
      -t "sensor/images/raw" \
      --cafile /var/run/certs/ca.crt \
      -D CONNECT authentication-method "K8S-SAT" \
      -D CONNECT authentication-data "$(cat /var/run/secrets/tokens/broker-sat)" \
      -D PUBLISH user-property __ts $(date +%s)000:0:df
  fi

  # Wait for 1 second before the next iteration
  sleep 1
done'

Überprüfen der Ausgabe

Abonnieren Sie in einem neuen Terminal das Ausgabethema:

kubectl exec --stdin --tty mqtt-client -n azure-iot-operations -- sh -c '
mosquitto_sub -h aio-broker -p 18883 -t "analytics/sensor/processed" --cafile /var/run/certs/ca.crt \
-D CONNECT authentication-method "K8S-SAT" \
-D CONNECT authentication-data "$(cat /var/run/secrets/tokens/broker-sat)"'

Die Ausgabe sieht wie im folgenden Beispiel aus:

{"temperature":[{"count":9,"max":2984.4444444444443,"min":248.33333333333337,"average":1849.6296296296296,"last":2612.222222222222,"unit":"C","overtemp":true}],"humidity":[{"count":10,"max":76.0,"min":30.0,"average":49.7,"last":38.0}],"object":[{"result":"milk can; broom; screwdriver; binoculars, field glasses, opera glasses; toy terrier"}]}
{"temperature":[{"count":10,"max":2490.5555555555557,"min":430.55555555555554,"average":1442.6666666666667,"last":1270.5555555555557,"unit":"C","overtemp":true}],"humidity":[{"count":9,"max":87.0,"min":34.0,"average":57.666666666666664,"last":42.0}],"object":[{"result":"broom; Saint Bernard, St Bernard; radiator"}]}

Hier enthält die Ausgabe die Temperatur- und Luftfeuchtigkeitsdaten sowie die erkannten Objekte in den Bildern.

Konfiguration von benutzerdefinierten Datenflussdiagrammen

Dieser Abschnitt enthält ausführliche Informationen zum Konfigurieren von Datenflussdiagrammen mit WASM-Modulen. Es umfasst alle Konfigurationsoptionen, Datenflussendpunkte und erweiterte Einstellungen.

Übersicht über Datenflussdiagramme

Ein Datenflussdiagramm definiert, wie Daten über WebAssembly-Module zur Verarbeitung fließen. Jedes Diagramm besteht aus:

• Modus, der steuert, ob das Diagramm aktiviert oder deaktiviert ist
• Profilreferenz, die mit einem Datenflussprofil verknüpft ist, das Skalierungs- und Ressourceneinstellungen definiert
• Datenträgerpersistenz, die optional beständigen Speicher für den Graph-Zustand ermöglicht
• Knoten, die die Quell-, Verarbeitungs- und Zielkomponenten definieren
• Knotenverbindungen, die angeben, wie Daten zwischen Knoten fließen

Moduskonfiguration

Die Mode-Eigenschaft bestimmt, ob das Datenflussdiagramm aktiv Daten verarbeitet. Legen Sie den Modus auf Enabled oder Disabled fest. Wenn deaktiviert, beendet das Diagramm die Verarbeitung von Daten, behält seine Konfiguration jedoch bei.

{
  "mode": "Enabled"
}

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2026-03-01' = {
  // ... other properties
  properties: {
    mode: 'Enabled'  // or 'Disabled'
    // ... other properties
  }
}

apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowGraph
metadata:
  name: my-dataflow-graph
  namespace: azure-iot-operations
spec:
  mode: Enabled  # or Disabled
  # ... other properties

Profilreferenz

Die Profilreferenz verbindet Ihr Datenflussdiagramm mit einem Datenflussprofil, das Skalierungseinstellungen, Instanzenanzahlen und Ressourcengrenzwerte definiert. Wenn Sie keinen Profilverweis angeben, müssen Sie stattdessen einen Kubernetes-Besitzerverweis verwenden. Die meisten Szenarien verwenden das Standardprofil, das Azure IoT Einsatz bereitstellt.

az iot ops dataflowgraph apply \
  --name $GRAPH_NAME \
  --instance $AIO_INSTANCE_NAME \
  --resource-group $RESOURCE_GROUP \
  --profile $PROFILE \
  --config-file graph.json

resource defaultDataflowProfile 'Microsoft.IoTOperations/instances/dataflowProfiles@2026-03-01' existing = {
  parent: aioInstance
  name: 'default'
}

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2026-03-01' = {
  parent: defaultDataflowProfile  // This establishes the profile relationship
  // ... other properties
}

apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowGraph
metadata:
  name: my-dataflow-graph
  namespace: azure-iot-operations
spec:
  profileRef: default  # Reference to the profile name
  # ... other properties

Anfordern der Datenträgerpersistenz

Die Persistenz der Anforderungsdiskette hilft Datenflussdiagrammen, den Zustand über Neustarts hinweg aufrechtzuerhalten. Wenn du diese Funktion aktivierst, kann der Graph den Verarbeitungszustand wiederherstellen, wenn der verbundene Broker neu startet. Dieses Feature ist nützlich für zustandsbehaftete Verarbeitungsszenarien, in denen der Verlust von Zwischendaten problematisch wäre. Wenn Sie die Persistenz des Anforderungsdatenträgers aktivieren, speichert der Broker die MQTT-Daten, z. B. Nachrichten in der Abonnentenwarteschlange, auf dem Datenträger. Dieser Ansatz stellt sicher, dass die Datenquelle Ihres Datenflusses während Stromausfällen oder Brokerneustarts keinen Datenverlust erlebt. Der Broker hält die optimale Leistung, weil man die Persistenz pro Datenfluss konfiguriert, sodass nur die Datenflüsse, die Persistenz benötigen, diese Funktion nutzen.

Der Datenflussgraph stellt diese Persistenzanfrage während des Abonnements unter Verwendung einer MQTTv5-Benutzereigenschaft ab. Dieses Feature funktioniert nur, wenn:

• Der Datenfluss verwendet den MQTT-Broker als Quelle (Quellknoten mit MQTT-Endpunkt)
• Der MQTT-Broker hat Persistenz aktiviert, wobei der dynamische Persistenzmodus auf den Datentyp festgelegt ist Enabled, z. B. Abonnentenqueues.

Mit dieser Konfiguration können MQTT-Clients wie Datenflussgraphen die Festplattenpersistenz für ihre Abonnements anfordern, indem sie MQTTv5-Benutzereigenschaften verwenden. Für die detaillierte Konfiguration der MQTT-Broker-Persistenz siehe Konfigurieren der MQTT-Broker-Persistenz.

Die Einstellung akzeptiert Enabled oder Disabled, mit Disabled der Standardeinstellung.

{
  "requestDiskPersistence": "Enabled"
}

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2026-03-01' = {
  // ... other properties
  properties: {
    requestDiskPersistence: 'Enabled'
    // ... other properties
  }
}

apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowGraph
metadata:
  name: my-dataflow-graph
  namespace: azure-iot-operations
spec:
  requestDiskPersistence: Enabled
  # ... other properties

Benennungsregeln und Grenzwerte

Datenflussressourcen und ihre Komponenten unterliegen Benennungsbeschränkungen auf unterschiedlichen Ebenen.

Das Datenflussdiagramm erzwingt auch die folgenden strukturellen Regeln:

• Keine doppelten Knotennamen: Jeder Knoten im Diagramm muss einen eindeutigen Namen haben.
• Gültige Verbindungstypen: Der Graph erlaubt nur die folgenden Knotenverbindungstypen: Quelle-zu-Graph, Quelle-zu-Ziel, Graph-zu-Graph und Graph-zu-Ziel-Verbindung.
• Keine Zyklen: Das Diagramm kann keine Zirkelverbindungen enthalten, die endlose Verarbeitungsschleifen erstellen.
• Keine Selbstschleifen: Ein Knoten kann keine Verbindung mit sich selbst herstellen.
• Kein Themaüberlappung: Wenn eine Quelle und ein Ziel denselben Endpunkt verwenden, können sich ihre MQTT-Themen nicht überlappen, wodurch eine endlose Nachrichtenschleife entsteht.

Knotenkonfiguration

Knoten sind die Bausteine eines Datenflussdiagramms. Jeder Knoten hat einen eindeutigen Namen innerhalb des Diagramms und führt eine bestimmte Funktion aus. Das Diagramm enthält drei Arten von Knoten:

Quellknoten

Quellknoten definieren, wo Daten in das Diagramm eingegeben werden. Sie stellen eine Verbindung mit Datenflussendpunkten her, die Daten von MQTT-Brokern oder Kafka-Themen empfangen. Jeder Quellknoten muss Folgendes angeben:

• Endpunktverweis, der auf einen konfigurierten Datenflussendpunkt verweist.
• Datenquellen als Liste der MQTT-Themen oder Kafka-Themen, die abonniert werden sollen
• Ressourcenverweis (optional), der auf eine Azure Device Registry-Ressource zur Schemaableitung verweist

Das Datenquellen-Array unterstützt das Abonnement mehrerer Themen, ohne die Endpunktkonfiguration zu ändern. Diese Flexibilität bedeutet, dass Sie Endpunkte über verschiedene Datenflüsse hinweg wiederverwenden können.

{
  "nodeType": "Source",
  "name": "sensor-data-source",
  "sourceSettings": {
    "endpointRef": "default",
    "dataSources": [
      "sensor/+/temperature",
      "sensor/+/humidity"
    ],
    "assetRef": "weather-station-asset"
  }
}

{
  nodeType: 'Source'
  name: 'sensor-data-source'
  sourceSettings: {
    endpointRef: 'default'  // References a data flow endpoint
    dataSources: [
      'sensor/+/temperature'  // MQTT topic with wildcard
      'sensor/+/humidity'
    ]
    assetRef: 'weather-station-asset'  // Optional asset reference
  }
}

- nodeType: Source
  name: sensor-data-source
  sourceSettings:
    endpointRef: default  # References a data flow endpoint
    dataSources:
      - sensor/+/temperature  # MQTT topic with wildcard
      - sensor/+/humidity
    assetRef: weather-station-asset  # Optional asset reference

Diagrammverarbeitungsknoten

Diagrammverarbeitungsknoten enthalten die WebAssembly-Module, die Daten transformieren. Diese Knoten rufen WASM-Artefakte aus Containerregistrierungen ab und führen sie mithilfe der angegebenen Konfigurationsparameter aus. Für jeden Diagrammknoten ist Folgendes erforderlich:

• Referenz zum Registrierungsendpunkt, die auf einen Registrierungsendpunkt verweist, um Artefakte zu pullen
• Artefaktspezifikation, die den Modulnamen und die Version definiert, die abgerufen werden soll
• Konfigurationsparameter als Schlüsselwertpaare, die an das WASM-Modul übergeben werden

Das Konfigurationsarray unterstützt die Anpassung des Modulverhaltens, ohne das WASM-Artefakt neu aufbauen zu müssen. Allgemeine Konfigurationsoptionen umfassen Verarbeitungsparameter, Schwellenwerte, Konvertierungseinstellungen und Featurekennzeichnungen.

{
  "nodeType": "Graph",
  "name": "temperature-processor",
  "graphSettings": {
    "registryEndpointRef": "my-acr-endpoint",
    "artifact": "temperature-converter:2.1.0",
    "configuration": [
      {
        "key": "temperature_lower_bound",
        "value": "-40"
      },
      {
        "key": "temperature_upper_bound",
        "value": "3422"
      }
    ]
  }
}

{
  nodeType: 'Graph'
  name: 'temperature-processor'
  graphSettings: {
    registryEndpointRef: 'my-acr-endpoint'
    artifact: 'temperature-converter:2.1.0'
    configuration: [
      {
        key: 'temperature_lower_bound'
        value: '-40'
      }
      {
        key: 'temperature_upper_bound'
        value: '3422'
      }
    ]
  }
}

- nodeType: Graph
  name: temperature-processor
  graphSettings:
    registryEndpointRef: my-acr-endpoint
    artifact: temperature-converter:2.1.0
    configuration:
      - key: temperature_lower_bound
        value: "-40"
      - key: temperature_upper_bound
        value: "3422"

Sie übergeben die Konfigurationsschlüssel-Wert-Paare zur Laufzeit an das WASM-Modul. Das Modul kann auf diese Werte zugreifen, um sein Verhalten anzupassen. Mit diesem Ansatz können Sie:

• Stellen Sie dasselbe WASM-Modul mit verschiedenen Konfigurationen bereit.
• Passen Sie Verarbeitungsparameter an, ohne Module neu zu erstellen.
• Aktivieren oder Deaktivieren von Features basierend auf bereitstellungsanforderungen.
• Legen Sie umgebungsspezifische Werte wie Schwellenwerte oder Endpunkte fest.

Zielknoten

Zielknoten definieren, wohin der Graph verarbeitete Daten schickt. Sie verbinden sich mit Datenflussendpunkten, die Daten an MQTT-Broker oder andere Systeme senden. Jeder Zielknoten gibt Folgendes an:

• Endpunktverweis, der auf einen konfigurierten Datenflussendpunkt verweist.
• Datenziel als spezifisches Thema, Pfad oder Speicherort für Ausgabedaten.
• Ausgabeschemaeinstellungen (optional), die serialisierungsformat und Schemaüberprüfung definieren.

{
  "nodeType": "Destination",
  "name": "output",
  "destinationSettings": {
    "endpointRef": "default",
    "dataDestination": "processed-data/temperature",
  }
}

{
  nodeType: 'Destination'
  name: 'output'
  destinationSettings: {
    endpointRef: 'default'
    dataDestination: 'processed-data/temperature'
  }
}

- nodeType: Destination
  name: output
  destinationSettings:
    endpointRef: default
    dataDestination: processed-data/temperature

Knotenverbindungen

Knotenverbindungen definieren den Datenflusspfad zwischen Knoten. Jede Verbindung gibt einen Quellknoten und zielknoten an, wodurch die Verarbeitungspipeline erstellt wird. Sie können optional ein Schema in die Verbindung einschließen. Das Modul empfängt das Schema bei der Initialisierung, was die Schema-Validierung wie in diesem Beispiel unterstützt.

"nodeConnections": [
  {
    "from": {
      "name": "sensor-data-source",
      "schema": {
        "schemaRef": "aio-sr://sensor-input-schema:1",
        "serializationFormat": "Json"
      }
    },
    "to": {
      "name": "temperature-processor"
    }
  },
  {
    "from": {
      "name": "temperature-processor"
    },
    "to": {
      "name": "output"
    }
  }
]

nodeConnections: [
  {
    from: {
      name: 'sensor-data-source'
      schema: {
        schemaRef: 'aio-sr://sensor-input-schema:1'
        serializationFormat: 'Json'
      }
    }
    to: {
      name: 'temperature-processor'
    }
  }
  {
    from: {
      name: 'temperature-processor'
    }
    to: {
      name: 'output'
    }
  }
]

nodeConnections:
  - from:
      name: sensor-data-source
      schema:
        schemaRef: aio-sr://sensor-input-schema:1
        serializationFormat: Json
    to:
      name: temperature-processor
  - from:
      name: temperature-processor
    to:
      name: output

Datenflussendpunkte

Datenflussdiagramme stellen über Datenflussendpunkte eine Verbindung mit externen Systemen her. Der Endpunkttyp bestimmt, ob Sie ihn als Quelle, Ziel oder beides verwenden können.

MQTT-Endpunkte

MQTT-Endpunkte können sowohl als Quellen als auch als Ziele dienen. Sie verbinden sich mit MQTT-Brokern, einschließlich:

• Azure IoT Einsatz lokaler MQTT-Broker (in jedem Datenfluss erforderlich)
• Azure Event Grid MQTT
• Benutzerdefinierte MQTT-Broker

Ausführliche Konfigurationsinformationen finden Sie unter Konfigurieren von MQTT-Datenflussendpunkten.

Kafka-Endpunkte

Kafka-Endpunkte können sowohl als Quellen als auch als Ziele dienen. Sie verbinden sich mit Kafka-kompatiblen Systemen, darunter:

• Azure Event Hubs (kafka-kompatibel)
• Apache Kafka Cluster
• Confluent Cloud

Ausführliche Konfigurationsinformationen finden Sie unter Configure Azure Event Hubs- und Kafka-Datenflussendpunkten.

Registrierungsendpunkte

Registrierungsendpunkte bieten Zugriff auf Containerregistrierungen zum Abrufen von WASM-Modulen und Graphdefinitionen. Sie werden nicht direkt im Datenfluss verwendet, sondern von Graphverarbeitungsknoten referenziert.

Ausführliche Konfigurationsinformationen finden Sie unter Konfigurieren von Registrierungsendpunkten.

Fehlerbehebung bei Datenflussgrafiken

Dieser Abschnitt enthält Tipps zur Problembehandlung bei häufig auftretenden Problemen beim Arbeiten mit Datenflussdiagrammen.

RegistryEndpoint nicht gefunden

Wenn das Datenflussdiagramm nicht gestartet werden kann und meldet, dass er den Registrierungsendpunkt nicht finden kann, überprüfen Sie Folgendes:

1. Der Name des Registry-Endpunkts muss übereinstimmen: Der registryEndpointRef Wert in Ihrem Datenflussdiagramm muss exakt mit dem name ihrer RegistryEndpoint Ressource übereinstimmen. Überprüfen Sie auf Tippfehler und die Beachtung der Groß- und Kleinschreibung.

   # List all registry endpoints in the namespace
   kubectl get registryendpoints -n azure-iot-operations
   

2. Der Registrierungsendpunkt befindet sich im richtigen Namespace: Der Registrierungsendpunkt muss sich im azure-iot-operations Namespace (oder demselben Namespace wie Ihr Datenflussdiagramm) befinden.

3. Der Registrierungsendpunkt ist bereit: Überprüfen Sie den Status Ihres Registrierungsendpunkts:

   kubectl describe registryendpoint $REGISTRY_ENDPOINT -n azure-iot-operations
   

4. Authentication ist ordnungsgemäß konfiguriert: Wenn Sie verwaltete Identität verwenden, stellen Sie sicher, dass die Azure IoT Einsatz Arc-Erweiterung über AcrPull Berechtigungen für die Registrierung verfügt. Wenn Sie die anonyme Authentifizierung mit einer öffentlichen Registrierung verwenden, überprüfen Sie, ob die Host-URL korrekt ist.

5. Artefakte sind in der Registrierung vorhanden: Stellen Sie sicher, dass die Graphdefinition und die WASM-Module, auf die in Ihrem Diagramm verwiesen wird, in den erwarteten Tags in der Registrierung verfügbar sind:

   # Check if artifacts exist (example with ORAS)
   oras manifest fetch $REGISTRY_HOST/graph-simple:1.0.0
   

Der Datenflussgraph wird ausgeführt, aber verarbeitet keine Daten.

Wenn Sie das Datenflussdiagramm bereitstellen, aber keine Nachrichten verarbeiten:

1. Überprüfen des Datenflussdiagrammstatus: Suchen Sie nach Fehlern im Datenflussdiagramm-Ressourcenstatus.

   kubectl get dataflowgraph $GRAPH_NAME -n azure-iot-operations -o yaml
   

2. Überprüfen Sie MQTT-Themen: Stellen Sie sicher, dass die Quellthemen in Ihrem Datenflussdiagramm den Themen entsprechen, in denen Sie Daten veröffentlichen.

3. Überprüfen von Zeitstempeln: Datenflussdiagramme verwenden zeitstempel (Hybrid Logical Clock, HLC) für die Nachrichtenverarbeitung. Fügen Sie die __ts Benutzereigenschaft ein, wenn Sie MQTT-Nachrichten veröffentlichen, um eine rechtzeitige Verarbeitung sicherzustellen.

Verwandte Inhalte

• Erstellen von WASM-Modulen für Datenflüsse für IDE- und CLI-basierte Entwicklungsworkflows
• Konfigurieren Sie WebAssembly-Graphdefinitionen für die YAML-Graphstruktur und Konfigurationsparameter
• Bereitstellen von WASM-Modulen und Graphdefinitionen für die Registrierungseinrichtung und artefaktverwaltung
• Führen Sie die ONNX-Inferenz in WASM für die Integration von maschinellen Lernmodellen aus
• Konfigurieren von MQTT-Datenflussendpunkten
• Configure Azure Event Hubs- und Kafka-Datenflussendpunkte
• Azure Data Lake Storage-Endpunkte für Datenflüsse konfigurieren
• Configure Microsoft Fabric OneLake-Datenflussendpunkte