Verwenden von WebAssembly (WASM) mit Datenflussdiagrammen

Azure IoT Operations-Datenflussdiagramme unterstützen WebAssembly(WASM)-Module für die benutzerdefinierte Datenverarbeitung am Edge. Sie können benutzerdefinierte Geschäftslogik und Datentransformationen als Teil Ihrer Datenflusspipelinen bereitstellen.

Voraussetzungen

• Stellen Sie eine Azure IoT Operations-Instanz auf einem Arc-fähigen Kubernetes-Cluster bereit. Weitere Informationen finden Sie unter Azure IoT-Betrieb bereitstellen.
• Konfigurieren Sie Ihre Containerregistrierung, und fügen Sie die Beispieldiagrammdefinitionen und WASM-Module hinzu, indem Sie die Anleitungen in "Deploy WebAssembly(WASM)"-Module und -Diagrammdefinitionen ausführen.

Überblick

Mit WebAssembly-Modulen (WASM) in Azure IoT Operations-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.

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:
   • Erstellen von WASM-Modulen für Datenflüsse in VS Code
   • Entwickeln von WebAssembly-Modulen (WASM)
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. Konfigurieren Sie Registrierungsendpunkte: Richten Sie Authentifizierungs- und Verbindungsdetails ein, damit Azure IoT Operations auf die Containerregistrierung zugreifen kann. Weitere Informationen finden Sie unter Konfigurieren von Registrierungsendpunkten.
5. Erstellen eines Datenflussdiagramms: Verwenden Sie die Nutzeroberfläche für Betriebsvorgänge oder Bicep-Dateien, um einen Datenfluss zu definieren, der eine Graphdefinition nutzt.
6. Bereitstellen und Ausführen: Azure IoT Operations 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 Temperaturmoduls 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.

Funktionsweise

Die Diagrammdefinition 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.

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-/Senkenvorgä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.

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@2025-10-01' existing = {
  name: aioInstanceName
}

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

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

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2025-10-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: '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: 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. 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)"'

Sie sehen die Temperaturdaten, die vom WASM-Modul von Fahrenheit in Celsius konvertiert wurden.

{"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 komplexe Diagrammdefinition koordiniert mehrere WASM-Module, um erweiterte Analysen und Objekterkennung durchzuführen.

Funktionsweise

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. 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.

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@2025-10-01' existing = {
  name: aioInstanceName
}

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

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

resource complexDataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2025-10-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: '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: 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. Sie befinden sich im Ordner images 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

Um RAW-Bilddateien aus dem Ordner ./samples/wasm/images in den Pod mqtt-client hochzuladen, können Sie den folgenden Befehl verwenden:

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 Luftfeuchtigkeitsdaten und Senden von Bildern

Sie können die Befehle zum Veröffentlichen von Temperatur- und Luftfeuchtigkeitsdaten und das Senden von Bildern in einem einzelnen Skript kombinieren. 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 Daten aktiv verarbeitet. Sie können den Modus auf Enabled oder Disabled (groß-/kleinschreibungsunabhängig) festlegen. Wenn deaktiviert, beendet das Diagramm die Verarbeitung von Daten, behält seine Konfiguration jedoch bei.

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2025-10-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 von Azure IoT Operations bereitgestellt wird.

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

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2025-10-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

Das Anfordern der Datenträgerpersistenz ermöglicht Datenflussdiagrammen, den Zustand über Neustarts hinweg beizubehalten. Wenn Sie dieses Feature aktivieren, kann das Diagramm den Verarbeitungszustand wiederherstellen, wenn der verbundene Broker neu gestartet wird. 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 behält eine optimale Leistung bei, da die Persistenz pro Datenfluss konfiguriert ist, sodass nur die Datenflüsse, die persistenz benötigen, dieses Feature verwenden.

Das Datenflussdiagramm macht diese Persistenzanforderung während des Abonnements mithilfe einer MQTTv5-Benutzereigenschaft. 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 Datenflussdiagramme Datenträgerpersistenz für ihre Abonnements mithilfe von MQTTv5-Benutzereigenschaften anfordern. 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.

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2025-10-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

Knotenkonfiguration

Knoten sind die Bausteine eines Datenflussdiagramms. Jeder Knoten hat einen eindeutigen Namen innerhalb des Diagramms und führt eine bestimmte Funktion aus. Es gibt 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:

• Endpunktreferenz, die auf einen konfigurierten Datenflussendpunkt verweist
• Datenquellen als Liste der MQTT-Themen oder Kafka-Themen, die abonniert werden sollen
• Ressourcenverweis (optional), der mit einem Azure Device Registry-Asset verknüpft ist, um Schemaableitung zu ermöglichen.

Mit dem Datenquellenarray können Sie mehrere Themen abonnieren, ohne die Endpunktkonfiguration zu ändern. Diese Flexibilität ermöglicht die Wiederverwendung von Endpunkten in verschiedenen Datenflüssen.

{
  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 mit angegebenen Konfigurationsparametern 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

Mit dem Konfigurationsarray können Sie das Modulverhalten anpassen, ohne das WASM-Artefakt neu zu erstellen. 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: 'input-unit'
        value: 'fahrenheit'
      }
      {
        key: 'output-unit' 
        value: 'celsius'
      }
      {
        key: 'precision'
        value: '2'
      }
      {
        key: 'enable-filtering'
        value: 'true'
      }
    ]
  }
}

- nodeType: Graph
  name: temperature-processor
  graphSettings:
    registryEndpointRef: my-acr-endpoint
    artifact: temperature-converter:2.1.0
    configuration:
      - key: input-unit
        value: fahrenheit
      - key: output-unit
        value: celsius
      - key: precision
        value: "2"
      - key: enable-filtering
        value: "true"

Die Konfigurationsschlüssel-Wert-Paare werden zur Laufzeit an das WASM-Modul übergeben. Das Modul kann auf diese Werte zugreifen, um sein Verhalten anzupassen. Dieser Ansatz ermöglicht Folgendes:

• Bereitstellen des gleichen WASM-Moduls mit unterschiedlichen Konfigurationen
• Anpassen von Verarbeitungsparametern ohne Neuerstellung von Modulen
• Aktivieren oder Deaktivieren von Features basierend auf Bereitstellungsanforderungen
• Festlegen von umgebungsspezifischen Werten wie Schwellenwerten oder Endpunkten

Zielknoten

Zielknoten definieren, wo verarbeitete Daten gesendet werden. Sie verbinden sich mit Datenflussendpunkten, die Daten an MQTT-Broker, Cloudspeicher oder andere Systeme senden. Jeder Zielknoten gibt Folgendes an:

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

Für Speicherziele wie Azure Data Lake oder Fabric OneLake können Sie Ausgabeschemaeinstellungen angeben, um zu steuern, wie Daten serialisiert und überprüft werden.

{
  nodeType: 'Destination'
  name: 'cloud-storage-destination'
  destinationSettings: {
    endpointRef: 'azure-storage-endpoint'
    dataDestination: 'processed-data/temperature'
    outputSchemaSettings: {
      serializationFormat: 'Parquet'
      schemaRef: 'temperature-output-schema:1'
    }
  }
}

- nodeType: Destination
  name: cloud-storage-destination
  destinationSettings:
    endpointRef: azure-storage-endpoint
    dataDestination: processed-data/temperature
    outputSchemaSettings:
      serializationFormat: Parquet
      schemaRef: temperature-output-schema:1

Knotenverbindungen

Knotenverbindungen definieren den Datenflusspfad zwischen Knoten. Jede Verbindung gibt einen Quellknoten und zielknoten an, wodurch die Verarbeitungspipeline erstellt wird. Verbindungen können optional die Schemaüberprüfung einschließen, um die Datenintegrität zwischen Verarbeitungsphasen sicherzustellen.

Wenn Sie die Schemaüberprüfung angeben, überprüft das System das Datenformat und die Struktur, während es zwischen Knoten fließt. Die Validierung hilft, Dateninkonsistenzen frühzeitig abzufangen und zu gewährleisten, dass WASM-Module Daten im erwarteten Format empfangen.

nodeConnections: [
  {
    from: {
      name: 'sensor-data-source'
      schema: {
        schemaRef: 'sensor-input-schema:1'
        serializationFormat: 'Json'
      }
    }
    to: {
      name: 'temperature-processor'
    }
  }
  {
    from: {
      name: 'temperature-processor'
    }
    to: {
      name: 'cloud-storage-destination'
    }
  }
]

nodeConnections:
  - from:
      name: sensor-data-source
      schema:
        schemaRef: sensor-input-schema:1
        serializationFormat: Json
    to:
      name: temperature-processor
  - from:
      name: temperature-processor
    to:
      name: cloud-storage-destination

Datenflussendpunkte

Datenflussdiagramme stellen über Datenflussendpunkte eine Verbindung mit externen Systemen her. Der Endpunkttyp bestimmt, ob er als Quelle, Ziel oder beides verwendet werden kann:

MQTT-Endpunkte

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

• Lokaler MQTT-Broker für Azure IoT Operations (erforderlich in jedem Datenfluss)
• 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 Konfigurieren von Azure Event Hubs und Kafka-Datenflussendpunkten.

Speicherendpunkte

Speicherendpunkte können nur als Ziele dienen. Sie verbinden sich mit Cloudspeichersystemen für langfristige Datenaufbewahrung und -analysen:

• Azure Data Lake-Speicher
• Microsoft Fabric OneLake
• Lokaler Speicher

Speicherendpunkte erfordern in der Regel Ausgabeschemaeinstellungen zum Definieren des Daten serialisierungsformats.

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.

Verwandte Inhalte

• Konfigurieren von MQTT-Datenflussendpunkten
• Konfigurieren von Azure Event Hubs und Kafka-Datenflussendpunkten
• Konfigurieren von Azure Data Lake Storage-Datenflussendpunkten
• Konfigurieren von Microsoft Fabric OneLake-Datenflussendpunkten