Gebruik WebAssembly (WASM) transformaties in datastroomgrafieken

Azure IoT-bewerkingen gegevensstroomgrafieken bevatten ingebouwde transformaties voor algemene verwerkingstaken, zoals toewijzing, filteren en aggregatie. Wanneer u aangepaste logica nodig hebt buiten wat de ingebouwde transformaties bieden, kunt u WEBAssembly-modules (WASM) implementeren als aangepaste transformaties in uw pijplijnen voor gegevensstroomgrafieken.

Belangrijk

Momenteel ondersteunt de operations experience webinterface alleen het creëren en bekijken van datastroomgrafiekartefacten afkomstig van Azure Container Registry (ACR) en, voor ingebouwde transformaties, mcr.microsoft.com. Voor meer informatie, zie Operations experience webUI toont alleen artefacten van datastroomgrafieken afkomstig van Azure Container Registry (ACR) en mcr.microsoft.com.

Vereiste voorwaarden

Een exemplaar van Azure IoT-bewerkingen geïmplementeerd in een Kubernetes-cluster. Zie Deploy Azure IoT-bewerkingen voor meer informatie.

Configureer een registereindpunt voor toegang tot WASM-modules en grafiekdefinities. U hebt twee opties:
- Snel aan de slag met een openbaar register: maak een registereindpunt voor ghcr.io anonieme verificatie. Zie Vooraf samengestelde modules uit een openbaar register gebruiken voor instructies.
- Privéregister: Stel uw eigen containerregister in en push de voorbeeldmodules door de richtlijnen in WASM-modules (Deploy WebAssembly) en grafiekdefinities te volgen.

De Azure CLI voorbeelden in dit artikel gebruiken omgevingsvariabelen zodat je elke waarde één keer kunt instellen en vervolgens de commando's kunt kopiëren en plakken as-is. Als je de Azure IoT-bewerkingen Codespaces-omgeving vanuit de quickstart gebruikt, zijn deze variabelen al voor je ingesteld en kun je deze stap overslaan. Anders stel je de volgende omgevingsvariabelen in je shell voordat je de commando's uitvoert.

De volgende scripts stellen de meest gebruikte omgevingsvariabelen in:

Omgevingsvariabele	Description
`SUBSCRIPTION_ID`	De ID van het abonnement dat je Azure IoT-bewerkingen-instantie bevat.
`RESOURCE_GROUP`	De naam van de resourcegroep die je Azure IoT-bewerkingen-instantie bevat.
`AIO_INSTANCE_NAME`	De naam van je Azure IoT-bewerkingen instance. Voer `az iot ops list -o table` uit om je instanties weer te geven.
`CLUSTER_NAME`	De naam van de Azure Arc-enabled Kubernetes-cluster die jouw instantie host.
`LOCATION`	De Azure-regio om te gebruiken voor nieuwe bronnen, bijvoorbeeld `eastus`.

Bash
PowerShell

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

Je hoeft alleen de variabelen in te stellen die dit artikel gebruikt. Dit artikel kan extra omgevingsvariabelen gebruiken voor de bronnamen die je kiest. Het artikel legt uit hoe je ze op de plek kunt plaatsen waar ze worden geïntroduceerd.

Dit artikel gebruikt ook de volgende omgevingsvariabelen voor de waarden die je kiest: REGISTRY_ENDPOINT (de naam van het register-endpoint), GRAPH_NAME (de naam van de datastroomgrafiek), PROFILE (de naam van het datastroomprofiel), en REGISTRY_HOST (de hostnaam van het containerregister). Stel elk commando in voordat je de bijbehorende commando's uitvoert.

Overzicht

Met behulp van WebAssembly-modules (WASM) in Azure IoT-bewerkingen gegevensstroomgrafieken kunt u gegevens aan de rand verwerken met hoge prestaties en beveiliging. WASM wordt uitgevoerd in een sandbox-omgeving en ondersteunt Rust en Python.

Een datastroom is een pijplijn die data tussen eindpunten verplaatst en transformeert door gebruik te maken van ingebouwde transformaties. Een gegevensstroomgrafiek breidt gegevensstromen uit met composeerbare verwerkingsstappen. Azure IoT-bewerkingen biedt ingebouwde gegevensstroomdiagrammen voor algemene bewerkingen, zoals toewijzing, filteren, vertakkingen en aggregatie. Voor aangepaste verwerkingslogica, implementeer WebAssembly-modules zoals dit artikel beschrijft. Gegevensstroomgrafieken maken gebruik van YAML-grafiekdefinities die aangeven hoe operators verbinding maken. De gegevensstroomgrafiekresource verpakt deze definitie en wijst de abstracte bron- en sinkbewerkingen toe aan concrete eindpunten, zoals MQTT-onderwerpen en Kafka-onderwerpen.

Aanbeveling

Voor de meeste scenario's voor gegevensverwerking begint u met de ingebouwde transformaties. Gebruik WASM-transformaties wanneer u aangepaste bedrijfslogica, gespecialiseerde algoritmen of verwerking nodig hebt die niet door de ingebouwde opties worden behandeld.

Belangrijk

Gegevensstroomgrafieken ondersteunen momenteel alleen MQTT-, Kafka- en OpenTelemetry-eindpunten. Andere eindpunttypen, zoals Data Lake, Microsoft Fabric OneLake, Azure Data Explorer en Lokale opslag, worden niet ondersteund.

Hoe WASM-gegevensstroomgrafieken werken

De implementatie van de WASM-gegevensstroom volgt deze werkstroom:

WASM-modules ontwikkelen: Schrijf aangepaste verwerkingslogica in een ondersteunde taal en compileer deze naar de webassembly-componentmodelindeling. Voor meer informatie, zie Build WASM-modules voor datastromen.
Grafiekdefinitie ontwikkelen: Definieer hoe gegevens door de modules worden verplaatst met behulp van YAML-configuratiebestanden. Zie WebAssembly-grafiekdefinities configureren voor meer informatie.
Artefacten opslaan in het register: push de gecompileerde WASM-modules en grafiekdefinities naar een containerregister met behulp van OCI-compatibele hulpprogramma's zoals ORAS. Zie WebAssembly-modules (WASM) en grafiekdefinities implementeren voor meer informatie.
Registereindpunten configureren: verificatie- en verbindingsgegevens instellen zodat Azure IoT-bewerkingen toegang heeft tot het containerregister. Zie Registereindpunten configureren voor meer informatie.
Gegevensstroomgrafiek maken: Gebruik de webinterface van de bewerkingservaring of Bicep bestanden om een gegevensstroom te definiëren die gebruikmaakt van een grafiekdefinitie.
Deploy and execute: Azure IoT-bewerkingen haalt grafiekdefinities en WASM-modules op uit het containerregister en voert deze uit.

In de volgende voorbeelden ziet u hoe u WASM-gegevensstroomgrafieken configureert voor veelvoorkomende scenario's. In de voorbeelden worden vastgelegde waarden en vereenvoudigde configuraties gebruikt, zodat u snel aan de slag kunt.

Voorbeeld 1: Basisimplementatie met één WASM-module

In dit voorbeeld worden temperatuurgegevens van Fahrenheit geconverteerd naar Celsius met behulp van een WASM-module. De broncode van de temperature-module is beschikbaar op GitHub. Als u de voorbeeldstappen in WebAssembly (WASM) modules en grafiekdefinities uitrollen hebt gevolgd, bevinden de graph-simple:1.0.0 grafiekdefinitie, en de vooraf gecompileerde temperature:1.0.0 module zich al in uw containerregister. Het pad van het voorbeeld-graphartefact is azure-samples/explore-iot-operations/graph-simple:1.0.0. Gebruik dit pad voor de openbare GHCR-voorbeelden of wanneer u de voorbeeldartefacten naar uw eigen register kopieert met hetzelfde opslagplaatspad.

Hoe de eenvoudige grafiek werkt

De definitie graph maakt een eenvoudige pijplijn met drie fasen:

Bron: ontvangt temperatuurgegevens van MQTT
Kaart: Verwerkt gegevens met de TEMPERATUUR WASM-module
Sink: hiermee worden geconverteerde gegevens terug naar MQTT verzonden

Zie voorbeeld 1: Eenvoudige grafiekdefinitie voor meer informatie over de werking van de eenvoudige grafiekdefinitie en de bijbehorende structuur.

Invoerindeling:

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

Uitvoerindeling:

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

Met de volgende configuratie maakt u een gegevensstroomgrafiek die gebruikmaakt van deze pijplijn voor temperatuurconversie. De gegevensstroomgrafiek verwijst naar de graph-simple:1.0.0 YAML-grafiekdefinitie en haalt de temperatuurmodule op uit uw containerregister. De Azure-voorbeeldgrafiek is opgeslagen onder het opslagplaatspad azure-samples/explore-iot-operations, dus neem dat pad op in de waarde van artifact.

De gegevensstroomgrafiek configureren

Deze configuratie definieert drie knooppunten die de werkstroom voor temperatuurconversie implementeren: een bronknooppunt dat zich abonneert op binnenkomende temperatuurgegevens, een knooppunt voor grafiekverwerking waarmee de WASM-module wordt uitgevoerd en een doelknooppunt dat de geconverteerde resultaten publiceert.

De gegevensstroomdiagramresource verpakt het artefact voor de gegevensstroomdefinitie en verbindt de abstracte bron- en sink-operaties met concrete eindpunten.

De source-bewerking van de grafiekdefinitie verbindt zich met het bronknooppunt van de gegevensstroom (MQTT-onderwerp)
De sink bewerking van de grafiekdefinitie maakt verbinding met het doelknooppunt van de gegevensstroom (MQTT-onderwerp)
De verwerkingsbewerkingen van de grafiekdefinitie worden uitgevoerd in het knooppunt voor graafverwerking

Met deze scheiding kunt u dezelfde grafiekdefinitie implementeren met verschillende eindpunten in verschillende omgevingen, terwijl de verwerkingslogica ongewijzigd blijft.

Als u een gegevensstroomgrafiek wilt maken in de bewerkingservaring, gaat u naar het tabblad Gegevensstroom .
Selecteer het dropdownmenu naast + Aanmaken en selecteer Create a data flow graph.
Selekteer de tijdelijke naam nieuwe gegevensstroom om de eigenschappen van de gegevensstroom in te stellen. Voer de naam van de gegevensstroomgrafiek in en kies het gegevensstroomprofiel dat u wilt gebruiken.

Selecteer bron in het gegevensstroomdiagram om het bronknooppunt te configureren. Selecteer onder Brondetailshet eindpunt van de asset of gegevensstroom.

Als u Asset selecteert, kiest u de asset waaruit u gegevens wilt ophalen en selecteert u Toepassen.

Als u Eindpunt voor gegevensstroom selecteert, voert u de volgende details in en selecteert u Toepassen.

Configuratie	Description
Eindpunt van gegevensstroom	Selecteer de standaardwaarde om het standaardeindpunt van de MQTT-berichtbroker te gebruiken.
Onderwerp	Het onderwerpfilter waarop u zich wilt abonneren voor binnenkomende berichten. Gebruik onderwerp(en)> Rij toevoegen om meerdere onderwerpen toe te voegen.
Berichtschema	Het schema dat moet worden gebruikt om de binnenkomende berichten te deserialiseren.

Selecteer in het gegevensstroomdiagram grafiektransformatie toevoegen (optioneel) om een knooppunt voor grafiekverwerking toe te voegen. Selecteer in het deelvenster Grafiekselectie, grafiek eenvoudig:1, en selecteer Toepassen.
U kunt bepaalde instellingen voor de grafiekoperator configureren door het grafiekknooppunt in het diagram te selecteren. U kunt bijvoorbeeld moduletemperatuur/kaartoperator selecteren en de waarde key2invoerenexample-value-2. Selecteer Toepassen om de wijzigingen op te slaan.
Selecteer Bestemming in het gegevensstroomdiagram om het doelknooppunt te configureren.
Selecteer Opslaan onder de naam van de gegevensstroomgrafiek om de gegevensstroomgrafiek op te slaan.

Opmerking

De artefactreferentie is relatief ten opzichte van de host van het registereindpunt. Voor de openbare GHCR-samples is de hostnaam van het registereindpunt ghcr.io, dus gebruik azure-samples/explore-iot-operations/graph-simple:1.0.0. Gebruik hetzelfde artefactpad als u de voorbeeldartefacten naar uw eigen register hebt gekopieerd onder hetzelfde opslagplaatspad. Gebruik voor uw eigen platte privéregisterindeling een verwijzing naar platte artefacten, zoals graph-simple:1.0.0.

De Azure CLI past een gegevensstroomgrafiek van één JSON-configuratiebestand toe. Maak een graph.json bestand met de grafiekeigenschappen:

{
  "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"
      }
    }
  ]
}

Pas het configuratiebestand toe. Stel de omgevingsvariabele GRAPH_NAME in op de naam van je datastroomgrafiek.

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'
        }
      }
    ]
  }
}

Belangrijk

Het gebruik van Kubernetes-implementatiemanifesten wordt niet ondersteund in productieomgevingen en mag alleen worden gebruikt voor foutopsporing en testen.

Maak een YAML-bestand met de volgende configuratie:

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

Sla de configuratie op als dataflow-graph.yaml. Pas dit vervolgens toe op uw cluster:

kubectl apply -f dataflow-graph.yaml

De gegevensstroom testen

Als u de gegevensstroom wilt testen, verzendt u MQTT-berichten vanuit het cluster. De temperatuurmodule verwacht berichten in een specifieke JSON-indeling met een genest temperature object dat value (numeriek) en unit (tekenreeks) velden bevat. Voorbeeld: {"temperature":{"value":72,"unit":"F"}}.

Implementeer eerst de MQTT-clientpod door de instructies te volgen in Testconnectiviteit met MQTT-broker met MQTT-clients. De MQTT-client biedt de verificatietokens en certificaten om verbinding te maken met de broker. Voer de volgende opdracht uit om de MQTT-client te implementeren:

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

Temperatuurberichten verzenden

Maak en voer in de eerste terminalsessie een script uit om temperatuurgegevens in Fahrenheit te verzenden:

# 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'

Opmerking

De MQTT-gebruikerseigenschap __ts voegt een tijdstempel toe aan de berichten om tijdige verwerking te garanderen door gebruik te maken van de Hybrid Logical Clock (HLC). De tijdstempel helpt de gegevensstroom te bepalen of het bericht moet worden geaccepteerd of weggevageld. De indeling van de eigenschap is <timestamp>:<counter>:<nodeid>. Het maakt de verwerking van de gegevensstroom nauwkeuriger, maar is niet verplicht.

Het script publiceert elke seconde willekeurige temperatuurgegevens naar het sensor/temperature/raw onderwerp. Dit moet er als volgt uitzien:

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

Laat het script actief om door te gaan met het publiceren van temperatuurgegevens.

Abonneer u in de tweede terminalsessie (ook verbonden met de MQTT-clientpod) op het uitvoeronderwerp om de geconverteerde temperatuurwaarden te bekijken:

# 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)"'

Je ziet dat de WASM-module de temperatuurgegevens omzet van Fahrenheit naar Celsius.

{"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}}

Voorbeeld 2: Een complexe grafiek implementeren

In dit voorbeeld ziet u een geavanceerde werkstroom voor gegevensverwerking die meerdere gegevenstypen verwerkt, waaronder temperatuur-, vochtigheids- en afbeeldingsgegevens. De complexe grafiekdefinitie organiseert meerdere WASM-modules om geavanceerde analyse- en objectdetectie uit te voeren.

Hoe de complexe grafiek werkt

De complexe grafiek verwerkt drie gegevensstromen en combineert deze in verrijkte sensoranalyses:

Temperatuurverwerking: Converteert Fahrenheit naar Celsius, filtert ongeldige metingen en berekent statistieken
Vochtigheidsverwerking: verzamelt vochtigheidsmetingen gedurende tijdsintervallen
Afbeeldingsverwerking: objectdetectie op cameramomentopnamen en indelingen van resultaten uitvoeren

Zie voorbeeld 2: Complexe grafiekdefinitie voor meer informatie over de werking van de complexe grafiekdefinitie, de structuur en de gegevensstroom door meerdere verwerkingsfasen.

De grafiek maakt gebruik van gespecialiseerde modules uit de verzameling Rust-operators.

De grafiek van de complexe gegevensstroom configureren

Met deze configuratie wordt de werkstroom voor multisensorverwerking geïmplementeerd met behulp van de graph-complex:1.0.0 YAML-grafiekdefinitie. Het pad van het voorbeeld-graphartefact is azure-samples/explore-iot-operations/graph-complex:1.0.0. U ziet hoe de implementatie van de gegevensstroomgrafiek vergelijkbaar is met voorbeeld 1 : beide gebruiken hetzelfde patroon met drie knooppunten (bron, grafiekprocessor, doel), ook al is de verwerkingslogica anders.

Deze overeenkomst treedt op omdat de gegevensstroomgrafiekresource fungeert als een hostomgeving waarmee grafiekdefinities worden geladen en uitgevoerd. De werkelijke verwerkingslogica bevindt zich in de grafiekdefinitie (graph-simple:1.0.0 of graph-complex:1.0.0), die de YAML-specificatie van bewerkingen en verbindingen tussen WASM-modules bevat. De gegevensstroomgrafiekresource biedt de runtime-infrastructuur voor het ophalen van de grafiekdefinitie, het instantiëren van de modules en het routeren van gegevens via de gedefinieerde werkstroom.

Als u een gegevensstroomgrafiek wilt maken in de bewerkingservaring, gaat u naar het tabblad Gegevensstroom .
Selecteer het dropdownmenu naast + Aanmaken en selecteer Create a data flow graph.
Selekteer de tijdelijke naam nieuwe gegevensstroom om de eigenschappen van de gegevensstroom in te stellen. Voer de naam van de gegevensstroomgrafiek in en kies het gegevensstroomprofiel dat u wilt gebruiken.

Selecteer bron in het gegevensstroomdiagram om het bronknooppunt te configureren. Selecteer onder Brondetailshet eindpunt van de asset of gegevensstroom.

Als u Asset selecteert, kiest u de asset waaruit u gegevens wilt ophalen en selecteert u Toepassen.

Als u Eindpunt voor gegevensstroom selecteert, voert u de volgende details in en selecteert u Toepassen.

Configuratie	Description
Eindpunt van gegevensstroom	Selecteer de standaardwaarde om het standaardeindpunt van de MQTT-berichtbroker te gebruiken.
Onderwerp	Het onderwerpfilter waarop u zich wilt abonneren voor binnenkomende berichten. Gebruik onderwerp(en)> Rij toevoegen om meerdere onderwerpen toe te voegen.
Berichtschema	Het schema dat moet worden gebruikt om de binnenkomende berichten te deserialiseren.

Selecteer in het gegevensstroomdiagram grafiektransformatie toevoegen (optioneel) om een knooppunt voor grafiekverwerking toe te voegen. Selecteer in het deelvenster Grafiekselectiegrafiekcomplex:1 en selecteer Toepassen.

Selecteer de grafiekknoop in het diagram om de instellingen voor graafoperatoren te configureren.

Operator	Description
module-snapshot/branch	Hiermee configureert u de `snapshot` module om objectdetectie uit te voeren op afbeeldingen. U kunt de `snapshot_topic` configuratiesleutel instellen om het invoeronderwerp voor afbeeldingsgegevens op te geven.
moduletemperatuur/kaart	Transformeert `key2` temperatuurwaarden naar een andere schaal.

Selecteer Toepassen om de wijzigingen op te slaan.
Selecteer Bestemming in het gegevensstroomdiagram om het doelknooppunt te configureren.
Selecteer Opslaan onder de naam van de gegevensstroomgrafiek om de gegevensstroomgrafiek op te slaan.

De Azure CLI past een gegevensstroomgrafiek van één JSON-configuratiebestand toe. Maak een graph.json bestand met de grafiekeigenschappen:

{
  "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"
      }
    }
  ]
}

Pas het configuratiebestand toe. De Azure CLI voegt automatisch toe extendedLocation vanuit de instantie- en resourcegroep, dus neem het niet in het bestand op.

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'
        }
      }
    ]
  }
}

Belangrijk

Het gebruik van Kubernetes-implementatiemanifesten wordt niet ondersteund in productieomgevingen en mag alleen worden gebruikt voor foutopsporing en testen.

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

De complexe gegevensstroom testen

Voordat u uitvoer kunt zien, stelt u de brongegevens in.

RAW-afbeeldingsbestanden uploaden naar de mqtt-clientpod

De afbeeldingsbestanden zijn bedoeld voor de snapshot module om objecten in de afbeeldingen te detecteren. De bestanden staan in de map afbeeldingen op GitHub.

Kloon eerst de opslagplaats om toegang te krijgen tot de afbeeldingsbestanden:

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

Gebruik de volgende opdracht om RAW-afbeeldingsbestanden van de ./samples/wasm/images map naar de mqtt-client pod te uploaden:

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

Controleer of de bestanden zijn geüpload:

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

U ziet nu de lijst met bestanden in de /tmp/images map.

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

Gesimuleerde temperatuur- en vochtigheidsgegevens publiceren en afbeeldingen verzenden

U kunt de opdrachten voor het publiceren van temperatuur- en vochtigheidsgegevens combineren en afbeeldingen naar één script verzenden. Gebruik de volgende opdracht:

# 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'

De uitvoer controleren

Abonneer u in een nieuwe terminal op het uitvoeronderwerp:

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)"'

De uitvoer ziet eruit als in het volgende voorbeeld:

{"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 bevat de uitvoer de temperatuur- en vochtigheidsgegevens, evenals de gedetecteerde objecten in de afbeeldingen.

Configuratie van aangepaste gegevensstroomgrafieken

Deze sectie bevat gedetailleerde informatie over het configureren van gegevensstroomgrafieken met WASM-modules. Hierin worden alle configuratieopties, gegevensstroomeindpunten en geavanceerde instellingen behandeld.

Overzicht van gegevensstroomgrafiek

Een gegevensstroomgrafiek definieert hoe gegevens stromen via WebAssembly-modules voor verwerking. Elke grafiek bestaat uit:

Modus waarmee wordt bepaald of de grafiek is ingeschakeld of uitgeschakeld
Profielreferentie die is gekoppeld aan een gegevensstroomprofiel dat schaalaanpassing en resource-instellingen definieert
Schijfpersistentie die optioneel permanente opslag voor grafiekstatus mogelijk maakt
Knooppunten die de bron-, verwerkings- en doelonderdelen definiëren
Knooppuntverbindingen die aangeven hoe gegevens stromen tussen knooppunten

Modusconfiguratie

De moduseigenschap bepaalt of de gegevensstroomgrafiek gegevens actief verwerkt. Stel de modus in op Enabled of Disabled (niet hoofdlettergevoelig). Wanneer deze is uitgeschakeld, stopt de grafiek met het verwerken van gegevens, maar blijft de configuratie behouden.

Wanneer u een gegevensstroomgrafiek maakt of bewerkt, gaat u in het deelvenster Eigenschappen van gegevensstroom naar Gegevensstroom inschakelen en selecteert u Ja om de modus in te stellen op Ingeschakeld. Als je het niet aanvinkt, is de modus uitgeschakeld.

Stel mode in op het hoogste niveau van je graph.json-configbestand en pas de graph toe met az iot ops dataflowgraph apply:

{
  "mode": "Enabled"
}

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

Belangrijk

Het gebruik van Kubernetes-implementatiemanifesten wordt niet ondersteund in productieomgevingen en mag alleen worden gebruikt voor foutopsporing en testen.

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

Profielreferentie

De profielreferentie verbindt uw gegevensstroomgrafiek met een gegevensstroomprofiel, waarmee schaalinstellingen, aantal exemplaren en resourcelimieten worden gedefinieerd. Als u geen profielreferentie opgeeft, moet u in plaats daarvan een Kubernetes-eigenaarsreferentie gebruiken. De meeste scenario's gebruiken het standaardprofiel dat Azure IoT-bewerkingen biedt.

Wanneer u een gegevensstroomgrafiek maakt of bewerkt, selecteert u in het deelvenster Eigenschappen van de gegevensstroom het gegevensstroomprofiel. De operations experience selecteert automatisch het standaard dataflowprofiel. Zie Gegevensstroomprofiel configureren voor meer informatie over gegevensstroomprofielen.

Belangrijk

Je kunt alleen het dataflowprofiel kiezen wanneer je een datastroomgrafiek maakt. Je kunt het dataflowprofiel niet aanpassen nadat je de dataflowgrafiek hebt gemaakt. Als u het gegevensstroomprofiel van een bestaande gegevensstroomgrafiek wilt wijzigen, verwijdert u de oorspronkelijke gegevensstroomgrafiek en maakt u een nieuw gegevensstroomprofiel met het nieuwe gegevensstroomprofiel.

Het profiel maakt geen deel uit van het graph.json configuratiebestand. Selecteer het in plaats daarvan door de --profile parameter te gebruiken wanneer je de graaf toepast met az iot ops dataflowgraph apply. Als u weglaat --profile, gebruikt de grafiek het default profiel:

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

Geef in Bicep het profiel op door als een subresource van het profiel de datastroomgrafiek te maken.

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
}

Belangrijk

Het gebruik van Kubernetes-implementatiemanifesten wordt niet ondersteund in productieomgevingen en mag alleen worden gebruikt voor foutopsporing en testen.

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

Schijfpersistentie aanvragen

Request disk persistence helpt datastroomgrafieken de status te behouden tijdens herstarts. Wanneer je deze functie inschakelt, kan de grafiek de verwerkingsstatus herstellen als de verbonden broker opnieuw opstart. Deze functie is handig voor staattoegepaste verwerkingsscenario's, waarbij verlies van tussenliggende gegevens problematisch zou zijn. Wanneer u schijfpersistentie aanvraagt, bewaart de broker de MQTT-gegevens, zoals berichten in de wachtrij voor abonnees, op schijf. Deze aanpak zorgt ervoor dat de gegevensbron van uw gegevensstroom geen gegevensverlies ondervindt tijdens stroomstoringen of dat broker opnieuw wordt opgestart. De broker behoudt optimale prestaties omdat je persistentie per datastroom configureert, dus alleen de datastromen die persistentie nodig hebben gebruiken deze functie.

De datastroomgrafiek doet dit persistentieverzoek tijdens het abonnement door gebruik te maken van een MQTTv5-gebruikerseigenschap. Deze functie werkt alleen wanneer:

De gegevensstroom gebruikt de MQTT-broker als bron (bronknooppunt met MQTT-eindpunt)
De MQTT-broker heeft persistentie ingeschakeld met dynamische persistentiemodus ingesteld op Enabled voor het gegevenstype, zoals abonneewachtrijen

Met deze configuratie kunnen MQTT-clients zoals datastroomgrafieken schijfpersistentie voor hun abonnementen aanvragen door gebruik te maken van MQTTv5-gebruikerseigenschappen. Zie MQTT-brokerpersistentie configureren voor gedetailleerde configuratie van MQTT-brokerpersistentie.

De instelling accepteert Enabled of Disabled, met Disabled als standaardwaarde.

Wanneer u een gegevensstroomgrafiek maakt of bewerkt, selecteert u in het deelvenster Gegevensstroomeigenschappen de optie Gegevenspersistentie aanvragen en kiest u Ja om de schijfpersistentie in te stellen op Ingeschakeld. Als u deze optie uitgeschakeld laat, is de instelling uitgeschakeld.

Stel requestDiskPersistence op het hoogste niveau van uw graph.json configuratiebestand in en pas de grafiek vervolgens toe met az iot ops dataflowgraph apply. De instelling accepteert Enabled of Disabled, met Disabled als standaardwaarde:

{
  "requestDiskPersistence": "Enabled"
}

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

Belangrijk

Het gebruik van Kubernetes-implementatiemanifesten wordt niet ondersteund in productieomgevingen en mag alleen worden gebruikt voor foutopsporing en testen.

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

Naamgevingsregels en -limieten

Resources voor gegevensstroomgrafieken en hun onderdelen hebben naamconventies die worden gehandhaafd op verschillende lagen.

Onderdeel	Toegestane tekens	Lengte	Aantekeningen
Resourcenaam voor gegevensstroomgrafiek	Kleine alfanumerieke tekens en koppeltekens (`a-z`, `0-9`, ). `-` Moet beginnen en eindigen met een alfanumerieke teken.	3-63 tekens	Afgedwongen door de Azure Resource Manager API.
Knooppuntnaam	Alfanumerieke tekens, onderstrepingstekens en afbreekstreepjes (`a-zA-Z0-9`, `_`, `-`).	Geen gedocumenteerde limiet	Moet uniek zijn binnen de grafiek.
Configuratiesleutel	Alfanumerieke tekens, onderstrepingstekens en afbreekstreepjes (`a-zA-Z0-9`, `_`, `-`).	Geen gedocumenteerde limiet	Sleutel-waardeparen die worden doorgegeven aan WASM-modules.
Naam van gegevensstroomprofiel	Kleine alfanumerieke letters en afbreekstreepjes.	3-39 tekens	De limiet van 39 tekens is te wijten aan kubernetes-podnaambeperkingen (limiet van 63 tekens min het `aio-dataflow-` voorvoegsel en revisieachtervoegsel).
Schemareferentie	Moet overeenkomen met de indeling `aio-sr://<namespace>/<name>:<version>` of `aio-sr://<name>:<version>`.	N/A	Wordt gebruikt in knooppuntverbindingsschema's.

De gegevensstroomgrafiek dwingt ook de volgende structurele regels af:

Geen dubbele knooppuntnamen: elk knooppunt in de grafiek moet een unieke naam hebben.
Geldige verbindingstypen: De grafiek staat alleen de volgende knooppuntverbindingen toe: Bron naar graaf, bron naar bestemming, graaf naar grafiek en graaf naar bestemming.
Geen cycli: de grafiek kan geen cirkelvormige verbindingen bevatten die oneindige verwerkingslussen zouden maken.
Geen zelflussen: een knooppunt kan geen verbinding maken met zichzelf.
Geen onderwerp overlapt: als een bron en doel hetzelfde eindpunt gebruiken, kunnen hun MQTT-onderwerpen niet overlappen, waardoor een oneindige berichtlus ontstaat.

Knooppuntconfiguratie

Knooppunten zijn de bouwstenen van een gegevensstroomgrafiek. Elk knooppunt heeft een unieke naam in de grafiek en voert een specifieke functie uit. De grafiek bevat drie typen knooppunten:

Bronknooppunten

Bronknooppunten definiëren waar gegevens de graf binnenkomen. Ze maken verbinding met eindpunten voor gegevensstromen die gegevens ontvangen van MQTT-brokers of Kafka-onderwerpen. Elk bronknooppunt moet het volgende opgeven:

Eindpuntreferentie die verwijst naar een geconfigureerd gegevensstroomeindpunt.
Gegevensbronnen als een lijst met MQTT-onderwerpen of Kafka-onderwerpen waarop u zich wilt abonneren
Assetreferentie (optioneel) die is gekoppeld aan een Azure Device Registry-asset voor schemadeductie

De databronnenarray ondersteunt het abonneren op meerdere onderwerpen zonder de eindconfiguratie aan te passen. Deze flexibiliteit betekent dat je eindpunten kunt hergebruiken over verschillende datastromen.

Opmerking

Momenteel ondersteunen datastroomgrafieken alleen MQTT- en Kafka-eindpunten als databronnen. Zie Eindpunten voor gegevensstromen configureren voor meer informatie.

Selecteer bron in het gegevensstroomdiagram om het bronknooppunt te configureren. Selecteer onder Brondetailshet eindpunt van de gegevensstroom en gebruik vervolgens het veld Onderwerp(en) om de MQTT-onderwerpfilters op te geven waarop u zich wilt abonneren voor binnenkomende berichten. Voeg meerdere MQTT-onderwerpen toe door rij toevoegen te selecteren en een nieuw onderwerp in te voeren.

De CLI past de hele grafiek uit één configuratiebestand toe, dus voeg dit bronknooppunt toe aan de nodes matrix in uw graph.json bestand en pas deze toe met az iot ops dataflowgraph apply:

{
  "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
  }
}

Belangrijk

Het gebruik van Kubernetes-implementatiemanifesten wordt niet ondersteund in productieomgevingen en mag alleen worden gebruikt voor foutopsporing en testen.

- 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

Knooppunten voor grafiekverwerking

Knooppunten voor grafiekverwerking bevatten de WebAssembly-modules waarmee gegevens worden getransformeerd. Deze knooppunten halen WASM-artefacten op uit containerregisters en voeren ze uit met behulp van opgegeven configuratieparameters. Voor elk grafiekknooppunt is het volgende vereist:

Registereindpuntverwijzing die verwijst naar een registereindpunt voor het ophalen van artefacten
Artefactspecificatie die de modulenaam en versie definieert die moet worden opgehaald
Configuratieparameters als sleutel-waardeparen die worden doorgegeven aan de WASM-module

De configuratie-array ondersteunt het aanpassen van modulegedrag zonder het WASM-artefact opnieuw op te bouwen. Algemene configuratieopties zijn onder andere verwerkingsparameters, drempelwaarden, conversie-instellingen en functievlagmen.

Selecteer in het gegevensstroomdiagram grafiektransformatie toevoegen (optioneel) om een knooppunt voor grafiekverwerking toe te voegen. Selecteer in het deelvenster Grafiekselectie het gewenste grafiekartefact, een eenvoudige of complexe grafiek en selecteer Toepassen. U kunt bepaalde instellingen voor de grafiekoperator configureren door het grafiekknooppunt in het diagram te selecteren.

De CLI past de hele grafiek uit één configuratiebestand toe, dus voeg dit grafiekverwerkingsknooppunt toe aan de nodes matrix in uw graph.json en pas deze toe met az iot ops dataflowgraph apply:

{
  "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'
      }
    ]
  }
}

Belangrijk

Het gebruik van Kubernetes-implementatiemanifesten wordt niet ondersteund in productieomgevingen en mag alleen worden gebruikt voor foutopsporing en testen.

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

U geeft de configuratiesleutel-waardeparen tijdens runtime door aan de WASM-module. De module heeft toegang tot deze waarden om het gedrag ervan aan te passen. Door deze aanpak te gebruiken, kunt u:

Implementeer dezelfde WASM-module met verschillende configuraties.
Pas de verwerkingsparameters aan zonder modules opnieuw te bouwen.
Functies in- of uitschakelen op basis van implementatievereisten.
Omgevingsspecifieke waarden instellen, zoals drempelwaarden of eindpunten.

Belangrijk

Raadpleeg de documentatie of broncode van uw WASM-module voor vereiste configuratieparameters. Als een module specifieke parameters verwacht (zoals filtergrenzen of drempels) en je die niet opgeeft, kan de module tijdens runtime falen. Zie Moduleconfiguratieparameters voor meer informatie over het definiëren van parameters in grafiekdefinities.

Doelknooppunten

Bestemmingsknooppunten bepalen waar de grafiek verwerkte data naartoe stuurt. Ze maken verbinding met eindpunten voor gegevensstromen die gegevens verzenden naar MQTT-brokers of andere systemen. Elk doelknooppunt geeft het volgende op:

Eindpuntreferentie die verwijst naar een geconfigureerd gegevensstroomeindpunt.
Gegevensbestemming als het specifieke onderwerp, pad of locatie voor uitvoergegevens.
Uitvoerschema-instellingen (optioneel) die de serialisatie-indeling en schemavalidatie definiëren.

Opmerking

Momenteel ondersteunen datastroomgrafieken alleen MQTT-, Kafka- en OpenTelemetrie-eindpunten als databestemmingen. Zie Eindpunten voor gegevensstromen configureren voor meer informatie.

Selecteer in het gegevensstroomdiagram het doelknooppunt .
Selecteer het gewenste gegevensstroomeindpunt in de vervolgkeuzelijst Details van gegevensstroomeindpunt .
Selecteer Doorgaan om de bestemming te configureren.
Voer de vereiste instellingen voor de bestemming in, inclusief het onderwerp of de tabel waarnaar de gegevens moeten worden verzonden. In de portal wordt het gegevensdoelveld automatisch geïnterpreteerd op basis van het eindpunttype. Als het eindpunt van de gegevensstroom bijvoorbeeld een MQTT-eindpunt is, wordt u op de pagina met doelgegevens gevraagd om het onderwerp in te voeren.

De CLI past de hele grafiek uit één configuratiebestand toe, dus voeg dit doelknooppunt toe aan de nodes matrix in uw graph.json en pas deze toe met az iot ops dataflowgraph apply:

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

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

Belangrijk

Het gebruik van Kubernetes-implementatiemanifesten wordt niet ondersteund in productieomgevingen en mag alleen worden gebruikt voor foutopsporing en testen.

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

Knooppuntverbindingen

Knooppuntverbindingen definiëren het gegevensstroompad tussen knooppunten. Elke verbinding geeft een bronknooppunt en doelknooppunt op, waardoor de verwerkingspijplijn wordt gemaakt. U kunt eventueel een schema opnemen in de verbinding. De module ontvangt het schema bij initialisatie, wat schemavalidatie ondersteunt zoals in dit voorbeeld.

De bewerkingservaring maakt automatisch knooppuntverbindingen wanneer u het knooppunt voor grafiekverwerking selecteert. Je kunt de verbindingen niet aanpassen nadat je de grafiek hebt gemaakt.

De CLI past de hele grafiek uit één configuratiebestand toe, dus voeg deze toe aan de nodeConnections matrix in uw graph.json bestand en pas deze toe met az iot ops dataflowgraph apply:

"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'
    }
  }
]

Belangrijk

Het gebruik van Kubernetes-implementatiemanifesten wordt niet ondersteund in productieomgevingen en mag alleen worden gebruikt voor foutopsporing en testen.

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

Eindpunten voor gegevensstroom

Gegevensstroomgrafieken maken verbinding met externe systemen via gegevensstroomeindpunten. Het type eindpunt bepaalt of u het kunt gebruiken als bron, doel of beide.

MQTT-eindpunten

MQTT-eindpunten kunnen fungeren als zowel bronnen als bestemmingen. Ze maken verbinding met MQTT-brokers, waaronder:

Azure IoT-bewerkingen lokale MQTT-broker (vereist in elke gegevensstroom)
Azure Event Grid MQTT
Aangepaste MQTT-brokers

Zie MQTT-gegevensstroomeindpunten configureren voor gedetailleerde configuratie-informatie.

Kafka-eindpunten

Kafka-eindpunten kunnen fungeren als zowel bronnen als bestemmingen. Ze maken verbinding met kafka-compatibele systemen, waaronder:

Azure Event Hubs (compatibel met Kafka)
Apache Kafka-clusters
Confluent Cloud

Zie Configure Azure Event Hubs- en Kafka-gegevensstroomeindpunten voor gedetailleerde configuratie-informatie.

Registersysteemeindpunten

Registereindpunten bieden toegang tot containerregisters voor het ophalen van WASM-modules en grafiekdefinities. Ze worden niet rechtstreeks in de gegevensstroom gebruikt, maar knooppunten voor grafiekverwerking verwijzen ernaar.

Zie Registereindpunten configureren voor gedetailleerde configuratie-informatie.

Problemen met gegevensstroomgrafieken oplossen

Deze sectie bevat tips voor het oplossen van veelvoorkomende problemen bij het werken met gegevensstroomgrafieken.

RegistryEndpoint is niet gevonden

Als de gegevensstroomgrafiek niet kan worden gestart en rapporteert dat het registereindpunt niet kan worden gevonden, controleert u het volgende:

De naam van het registryEndpointRef: de name waarde in de gegevensstroomgrafiek moet exact overeenkomen met de RegistryEndpoint resource. Controleer op typfouten en hoofdlettergevoeligheid.
```
# List all registry endpoints in the namespace
kubectl get registryendpoints -n azure-iot-operations
```
Het registereindpunt bevindt zich in de juiste naamruimte: het registereindpunt moet zich in de azure-iot-operations naamruimte bevinden (of dezelfde naamruimte als uw gegevensstroomgrafiek).

Het registereindpunt is gereed: controleer de status van uw registereindpunt:

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

Authenticatie is correct geconfigureerd: Als u beheerde identiteit gebruikt, zorg ervoor dat de Azure IoT-bewerkingen Arc-extensie AcrPull machtigingen voor het register heeft. Als u anonieme verificatie met een openbaar register gebruikt, controleert u of de host-URL juist is.
Artefacten bevinden zich in het register: controleer of de grafiekdefinitie en WASM-modules waarnaar in uw grafiek wordt verwezen, beschikbaar zijn op de verwachte tags in het register:
```
# Check if artifacts exist (example with ORAS)
oras manifest fetch $REGISTRY_HOST/graph-simple:1.0.0
```

Gegevensstroomgrafiek wordt uitgevoerd, maar verwerkt geen gegevens

Als u de gegevensstroomgrafiek implementeert, maar geen berichten verwerkt:

Controleer de grafiekstatus van de gegevensstroom: zoek naar fouten in de resourcestatus van de gegevensstroomgrafiek.
```
kubectl get dataflowgraph $GRAPH_NAME -n azure-iot-operations -o yaml
```
Controleer MQTT-onderwerpen: Zorg ervoor dat de brononderwerpen in uw gegevensstroomgrafiek overeenkomen met de onderwerpen waarin u gegevens publiceert.
Tijdstempels controleren: gegevensstroomgrafieken gebruiken HLC-tijdstempels (Hybrid Logical Clock) voor berichtverwerking. Voeg de __ts gebruikerseigenschap toe wanneer je MQTT-berichten publiceert om tijdige verwerking te garanderen.

WASM-modules bouwen voor gegevensstromen voor zowel IDE- als CLI-ontwikkelwerkstromen
WebAssembly-grafiekdefinities configureren voor de YAML-structuur en configuratieparameters van de grafiek
WASM-modules en grafiekdefinities implementeren voor registerinstallatie en artefactbeheer
Voer ONNX-inferentie uit in WASM voor machine learning-modelintegratie
MQTT-gegevensstroomeindpunten configureren
Configuratie van Azure Event Hubs- en Kafka-gegevensstroomeindpunten
Configureer Azure Data Lake Storage gegevensstroom-eindpunten
Configureer Microsoft Fabric OneLake-gegevensstroomeindpunten

Feedback

Is deze pagina nuttig?

Last updated on 2026-08-21

Gebruik WebAssembly (WASM) transformaties in datastroomgrafieken

Vereiste voorwaarden

Overzicht

Hoe WASM-gegevensstroomgrafieken werken

Voorbeeld 1: Basisimplementatie met één WASM-module

Hoe de eenvoudige grafiek werkt

De gegevensstroomgrafiek configureren

De gegevensstroom testen

Temperatuurberichten verzenden

Abonneren op verwerkte berichten

Voorbeeld 2: Een complexe grafiek implementeren

Hoe de complexe grafiek werkt

De grafiek van de complexe gegevensstroom configureren

De complexe gegevensstroom testen

RAW-afbeeldingsbestanden uploaden naar de mqtt-clientpod

Gesimuleerde temperatuur- en vochtigheidsgegevens publiceren en afbeeldingen verzenden

De uitvoer controleren

Configuratie van aangepaste gegevensstroomgrafieken

Overzicht van gegevensstroomgrafiek

Modusconfiguratie

Profielreferentie

Schijfpersistentie aanvragen

Naamgevingsregels en -limieten

Knooppuntconfiguratie

Bronknooppunten

Knooppunten voor grafiekverwerking

Doelknooppunten

Knooppuntverbindingen

Eindpunten voor gegevensstroom

MQTT-eindpunten

Kafka-eindpunten

Registersysteemeindpunten

Problemen met gegevensstroomgrafieken oplossen

RegistryEndpoint is niet gevonden

Gegevensstroomgrafiek wordt uitgevoerd, maar verwerkt geen gegevens

Verwante inhoud

Feedback

Aanvullende resources