Använda WebAssembly (WASM) med dataflödesdiagram (förhandsversion)

Azure IoT Operations dataflödesdiagram stöder WebAssembly-moduler (WASM) för anpassad databearbetning vid gränsen. Du kan distribuera anpassad affärslogik och datatransformeringar som en del av dina dataflödespipelines.

Förutsättningar

• Distribuera en Azure IoT Operations-instans, version 1.2 förhandsversion eller senare, i ett Arc-aktiverat Kubernetes-kluster. Mer information finns i Distribuera Azure IoT-åtgärder.
• Använd Azure Container Registry (ACR) för att lagra WASM-moduler och diagram.
• Installera OCI Registry As Storage (ORAS) CLI för att skicka WASM-moduler till registret.
• Utveckla anpassade WASM-moduler genom att följa vägledningen i Utveckla WebAssembly-moduler för dataflödesdiagram.

Översikt

Med WASM-moduler (WebAssembly) i Azure IoT Operations-dataflödesdiagram kan du bearbeta data vid gränsen med hög prestanda och säkerhet. WASM körs i en sandbox-miljö och har stöd för Rust och Python.

Så här fungerar WASM-dataflödesdiagram

Implementeringen av WASM-dataflödet följer det här arbetsflödet:

1. Utveckla WASM-moduler: Skriv anpassad bearbetningslogik på ett språk som stöds och kompilera den till formatet WebAssembly Component Model.
2. Utveckla grafdefinition: Definiera hur data flyttas genom modulerna med hjälp av YAML-konfigurationsfiler. Detaljerad information finns i Konfigurera WebAssembly-grafdefinitioner.
3. Lagra artefakter i registret: Skicka de kompilerade WASM-modulerna till ett containerregister med hjälp av OCI-kompatibla verktyg som ORAS.
4. Konfigurera registerslutpunkter: Konfigurera autentiserings- och anslutningsinformation så att Azure IoT Operations kan komma åt containerregistret.
5. Skapa dataflöde: Definiera datakällor, artefaktnamnet och mål.
6. Distribuera och köra: Azure IoT Operations hämtar WASM-moduler från registret och kör dem baserat på grafdefinitionen.

Kom igång med exempel

De här exemplen visar hur du konfigurerar och distribuerar WASM-dataflödesdiagram för vanliga scenarier. Exemplen använder hårdkodade värden och förenklade konfigurationer så att du snabbt kan komma igång.

Konfigurera containerregister

Azure IoT Operations behöver ett containerregister för att hämta WASM-moduler och grafdefinitioner. Du kan använda Azure Container Registry (ACR) eller ett annat OCI-kompatibelt register.

Information om hur du skapar och konfigurerar ett Azure Container Registry finns i Distribuera Azure Container Registry.

Installera ORAS CLI

Använd ORAS CLI för att push-överföra WASM-moduler och grafdefinitioner till containerregistret. Installationsinstruktioner finns i Installera ORAS.

Hämta exempelmoduler från det offentliga registret

I den här förhandsversionen använder du fördefinierade exempelmoduler:

# Pull sample modules and graphs
oras pull ghcr.io/azure-samples/explore-iot-operations/graph-simple:1.0.0
oras pull ghcr.io/azure-samples/explore-iot-operations/graph-complex:1.0.0
oras pull ghcr.io/azure-samples/explore-iot-operations/temperature:1.0.0
oras pull ghcr.io/azure-samples/explore-iot-operations/window:1.0.0
oras pull ghcr.io/azure-samples/explore-iot-operations/snapshot:1.0.0
oras pull ghcr.io/azure-samples/explore-iot-operations/format:1.0.0
oras pull ghcr.io/azure-samples/explore-iot-operations/humidity:1.0.0
oras pull ghcr.io/azure-samples/explore-iot-operations/collection:1.0.0
oras pull ghcr.io/azure-samples/explore-iot-operations/enrichment:1.0.0
oras pull ghcr.io/azure-samples/explore-iot-operations/filter:1.0.0

Skicka moduler till registret

När du har exempelmoduler och grafer skickar du dem till containerregistret. Ersätt <YOUR_ACR_NAME> med namnet på ditt Azure Container Registry.

# Log in to your ACR
az acr login --name <YOUR_ACR_NAME>

# Push modules to your registry
oras push <YOUR_ACR_NAME>.azurecr.io/graph-simple:1.0.0 graph-simple.yaml
oras push <YOUR_ACR_NAME>.azurecr.io/graph-complex:1.0.0 graph-complex.yaml
oras push <YOUR_ACR_NAME>.azurecr.io/temperature:1.0.0 temperature-1.0.0.wasm
oras push <YOUR_ACR_NAME>.azurecr.io/window:1.0.0 window-1.0.0.wasm
oras push <YOUR_ACR_NAME>.azurecr.io/snapshot:1.0.0 snapshot-1.0.0.wasm
oras push <YOUR_ACR_NAME>.azurecr.io/format:1.0.0 format-1.0.0.wasm
oras push <YOUR_ACR_NAME>.azurecr.io/humidity:1.0.0 humidity-1.0.0.wasm
oras push <YOUR_ACR_NAME>.azurecr.io/collection:1.0.0 collection-1.0.0.wasm
oras push <YOUR_ACR_NAME>.azurecr.io/enrichment:1.0.0 enrichment-1.0.0.wasm
oras push <YOUR_ACR_NAME>.azurecr.io/filter:1.0.0 filter-1.0.0.wasm

Skapa en registerslutpunkt

En registerslutpunkt definierar anslutningen till containerregistret. Dataflödesdiagram använder registerslutpunkter för att hämta WASM-moduler och grafdefinitioner från containerregister. Detaljerad information om hur du konfigurerar registerslutpunkter med olika autentiseringsmetoder och registertyper finns i Konfigurera registerslutpunkter.

För snabb installation med Azure Container Registry skapar du en registerslutpunkt med systemtilldelad hanterad identitetsautentisering:

param aioInstanceName string = '<AIO_INSTANCE_NAME>'
param customLocationName string = '<CUSTOM_LOCATION_NAME>'
param registryEndpointName string = '<REGISTRY_ENDPOINT_NAME>'
param acrName string = '<YOUR_ACR_NAME>'

resource aioInstance 'Microsoft.IoTOperations/instances@2025-07-01-preview'' existing = {
  name: aioInstanceName
}

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

resource registryEndpoint 'Microsoft.IoTOperations/instances/registryEndpoints@2025-07-01-preview'' = {
  parent: aioInstance
  name: registryEndpointName
  extendedLocation: {
    name: customLocation.id
    type: 'CustomLocation'
  }
  properties: {
    host: '${acrName}.azurecr.io'
    authentication: {
      method: 'SystemAssignedManagedIdentity'
      systemAssignedManagedIdentitySettings: {
        audience: 'https://management.azure.com/'
      }
    }
  }
}

apiVersion: connectivity.iotoperations.azure.com/v1beta1
kind: RegistryEndpoint
metadata:
  name: <REGISTRY_ENDPOINT_NAME>
  namespace: azure-iot-operations
spec:
  host: <YOUR_ACR_NAME>.azurecr.io
  authentication:
    method: SystemAssignedManagedIdentity
    systemAssignedManagedIdentitySettings:
      audience: https://management.azure.com/

Hämta tilläggsnamn

# Get extension name
az k8s-extension list \
  --resource-group <RESOURCE_GROUP> \
  --cluster-name <CLUSTER_NAME> \
  --cluster-type connectedClusters \
  --query "[?extensionType=='microsoft.iotoperations'].name" \
  --output tsv

Det första kommandot returnerar tilläggsnamnet (till exempel azure-iot-operations-4gh3y).

Konfigurera behörigheter för hanterad identitet

Om du vill låta Azure IoT Operations hämta WASM-moduler från containerregistret ger du den hanterade identiteten rätt behörigheter. IoT Operations-tillägget använder en systemtilldelad hanterad identitet som behöver rollen i AcrPull Azure Container Registry. Kontrollera att du har följande förutsättningar:

• Ägarbehörigheter för Azure Container Registry.
• Containerregistret kan finnas i en annan resursgrupp eller prenumeration, men det måste finnas i samma klientorganisation som din IoT Operations-distribution.

Kör dessa kommandon för att tilldela AcrPull rollen till den hanterade IoT Operations-identiteten:

# Get the IoT Operations extension managed identity
export EXTENSION_OBJ_ID=$(az k8s-extension list --cluster-name $CLUSTER_NAME -g $RESOURCE_GROUP --cluster-type connectedClusters --query "[?extensionType=='microsoft.iotoperations'].identity.principalId" -o tsv)

# Get the application ID for the managed identity
export SYSTEM_ASSIGNED_MAN_ID=$(az ad sp show --id $EXTENSION_OBJ_ID --query "appId" -o tsv)

# Assign the AcrPull role to the managed identity
az role assignment create --role "AcrPull" --assignee $SYSTEM_ASSIGNED_MAN_ID --scope "/subscriptions/$SUBSCRIPTION_ID/resourceGroups/$RESOURCE_GROUP/providers/Microsoft.ContainerRegistry/registries/$ACR_NAME"

Mer information om containerregisterroller finns i Roller och behörigheter för Azure Container Registry.

Om du får autentiseringsfel med Azure CLI tilldelar du behörigheter i Azure-portalen:

1. Gå till Azure Container Registry i Azure-portalen.
2. Välj Åtkomstkontroll (IAM) på menyn.
3. Välj Lägg till>Lägg till rolltilldelning.
4. Välj den inbyggda rollen AcrPull .
5. Välj Användare, grupp eller tjänstens huvudnamn som alternativet tilldela åtkomst till.
6. Sök efter och välj namnet på IoT Operations-tillägget (till exempel azure-iot-operations-4gh3y).
7. Välj Spara för att slutföra rolltilldelningen.

Detaljerade anvisningar finns i Tilldela Azure-roller med hjälp av Azure-portalen.

Exempel 1: Grundläggande distribution med en WASM-modul

Det här exemplet konverterar temperaturdata från Fahrenheit till Celsius med hjälp av en WASM-modul. Källkoden för temperaturmodulen är tillgänglig på GitHub. Använd den förkompilerade version graph-simple:1.0.0 som du skickade till containerregistret.

Så här fungerar det

Grafdefinitionen skapar en enkel pipeline i tre steg:

1. Källa: Tar emot temperaturdata från MQTT
2. Karta: Bearbetar data med wasm-modulen temperature
3. Mottagare: Skickar konverterade data tillbaka till MQTT

Detaljerad information om hur den enkla grafdefinitionen fungerar och dess struktur finns i Exempel 1: Enkel grafdefinition.

Indataformat:

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

Utdataformat:

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

Följande konfiguration skapar ett dataflödesdiagram som använder den här temperaturkonverteringspipelinen. Diagrammet refererar till graph-simple:1.0.0 artefakten, som innehåller YAML-definitionen och hämtar temperaturmodulen från containerregistret.

Konfigurera dataflödesdiagrammet

Den här konfigurationen definierar tre noder som implementerar arbetsflödet för temperaturkonvertering: en källnod som prenumererar på inkommande temperaturdata, en grafbearbetningsnod som kör WASM-modulen och en målnod som publicerar de konverterade resultaten.

Dataflödesdiagramresursen "omsluter" diagramdefinitionsartefakten och ansluter dess abstrakta käll-/mottagaråtgärder till konkreta slutpunkter:

• Grafdefinitionens source åtgärd ansluter till dataflödets källnod (MQTT-ämne)
• Diagramdefinitionens sink åtgärd ansluter till dataflödets målnod (MQTT-ämne)
• Diagramdefinitionens bearbetningsåtgärder körs i grafbearbetningsnoden

Med den här separationen kan du distribuera samma grafdefinition med olika slutpunkter mellan miljöer samtidigt som bearbetningslogik hålls oförändrad.

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-07-01-preview'' existing = {
  name: aioInstanceName
}

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

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

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2025-07-01-preview'' = {
  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/v1beta1
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

Testa dataflödet

Om du vill testa dataflödet skickar du MQTT-meddelanden inifrån klustret. Distribuera först MQTT-klientpodden genom att följa anvisningarna i Testa anslutningen till MQTT-koordinator med MQTT-klienter. MQTT-klienten tillhandahåller autentiseringstoken och certifikat för att ansluta till asynkron meddelandekö. Kör följande kommando för att distribuera MQTT-klienten:

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

Skicka temperaturmeddelanden

I den första terminalsessionen skapar och kör du ett skript för att skicka temperaturdata i 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'

Skriptet publicerar slumpmässiga temperaturdata till ämnet sensor/temperature/raw varje sekund. Det bör se ut så här:

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

Låt skriptet köras för att fortsätta publicera temperaturdata.

Prenumerera på bearbetade meddelanden

I den andra terminalsessionen (även ansluten till MQTT-klientpodden) prenumererar du på utdataavsnittet för att se de konverterade temperaturvärdena:

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

Du ser temperaturdata som konverterats från Fahrenheit till Celsius av WASM-modulen.

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

Exempel 2: Distribuera ett komplext diagram

Det här exemplet visar ett avancerat arbetsflöde för databearbetning som hanterar flera datatyper, inklusive temperatur, luftfuktighet och bilddata. Den komplexa grafdefinitionen samordnar flera WASM-moduler för att utföra avancerad analys och objektidentifiering.

Så här fungerar det

Det komplexa diagrammet bearbetar tre dataströmmar och kombinerar dem till berikad sensoranalys:

• Temperaturbearbetning: Konverterar Fahrenheit till Celsius, filtrerar ogiltiga avläsningar och beräknar statistik
• Fuktighetsbearbetning: Ackumulerar fuktighetsmätningar över tidsintervall
• Bildbearbetning: Utför objektidentifiering på kameraögonblicksbilder och formaterar resultat

Detaljerad information om hur den komplexa grafdefinitionen fungerar, dess struktur och dataflödet genom flera bearbetningssteg finns i Exempel 2: Komplex grafdefinition.

Diagrammet använder specialiserade moduler från Rust-exemplen.

Konfigurera det komplexa dataflödesdiagrammet

Den här konfigurationen implementerar arbetsflödet för bearbetning av flera sensorer med hjälp av graph-complex:1.0.0 artefakten. Observera hur dataflödesdiagramdistributionen liknar exempel 1 – båda använder samma mönster med tre noder (källa, grafprocessor, mål) även om bearbetningslogik skiljer sig.

Den här likheten beror på att dataflödesgrafresursen fungerar som en värdmiljö som läser in och kör grafdefinitioner. Den faktiska bearbetningslogik finns i diagramdefinitionsartefakten (graph-simple:1.0.0 vs graph-complex:1.0.0), som innehåller YAML-specifikationen för åtgärder och anslutningar mellan WASM-moduler. Dataflödesdiagramresursen tillhandahåller körningsinfrastrukturen för att hämta artefakten, instansiera modulerna och dirigera data via det definierade arbetsflödet.

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-07-01-preview'' existing = {
  name: aioInstanceName
}

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

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

resource complexDataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2025-07-01-preview'' = {
  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/v1beta1
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

Testa det komplexa dataflödet

Innan vi kan se utdata måste vi hämta konfigurationen av källdata.

Ladda upp RAW-avbildningsfiler till mqtt-klientpodden

Bildfilerna är till för modulen snapshot för att identifiera objekt i bilderna. De finns i avbildningsmappen på GitHub.

Klona först lagringsplatsen för att få åtkomst till avbildningsfilerna:

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

Om du vill ladda upp RAW-avbildningsfiler från ./samples/wasm/images mappen till mqtt-client podden kan du använda följande kommando:

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

Kontrollera att filerna har laddats upp:

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

Du bör se listan över filer i /tmp/images mappen.

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

Publicera simulerade temperatur-, fuktighetsdata och skicka bilder

Du kan kombinera kommandona för att publicera temperatur, fuktighetsdata och skicka bilder till ett enda skript. Använd följande kommando:

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

Kontrollera utdata

I en ny terminal prenumererar du på utdataavsnittet:

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

Utdata bör se ut så här

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

Här innehåller utdata temperatur- och luftfuktighetsdata samt identifierade objekt i bilderna.

Utveckla anpassade WASM-moduler

Om du vill skapa anpassad databearbetningslogik för dina dataflödesdiagram utvecklar du WebAssembly-moduler i Rust eller Python. Med anpassade moduler kan du implementera specialiserad affärslogik, datatransformering och analys som inte är tillgängliga i de inbyggda operatorerna.

För omfattande utvecklingsvägledning, inklusive:

• Konfigurera utvecklingsmiljön
• Skapa operatorer i Rust och Python
• Förstå datamodellen och gränssnitten
• Skapa och testa dina moduler

Se Utveckla WebAssembly-moduler för dataflödesdiagram.

Detaljerad information om hur du skapar och konfigurerar YAML-grafdefinitioner som definierar dina arbetsflöden för databearbetning finns i Konfigurera WebAssembly-grafdefinitioner.

Konfiguration av anpassade dataflödesdiagram

Det här avsnittet innehåller detaljerad information om hur du konfigurerar dataflödesdiagram med WASM-moduler. Den omfattar alla konfigurationsalternativ, dataflödesslutpunkter och avancerade inställningar.

Översikt över dataflödesdiagram

Ett dataflödesdiagram definierar hur data flödar via WebAssembly-moduler för bearbetning. Varje diagram består av:

• Läge som styr om grafen är aktiverad eller inaktiverad
• Profilreferens som länkar till en dataflödesprofil som definierar skalnings- och resursinställningar
• Diskpersistence som om du vill aktivera beständig lagring för graftillstånd
• Noder som definierar käll-, bearbetnings- och målkomponenter
• Nodanslutningar som anger hur data flödar mellan noder

Lägeskonfiguration

Lägesegenskapen avgör om dataflödesdiagrammet aktivt bearbetar data. Du kan ange läget till Enabled eller Disabled (skiftlägesokänsligt). När det är inaktiverat slutar grafen att bearbeta data men behåller sin konfiguration.

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

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

Profilreferens

Profilreferensen ansluter dataflödesdiagrammet till en dataflödesprofil som definierar skalningsinställningar, antal instanser och resursgränser. Om du inte anger någon profilreferens måste du använda en Kubernetes-ägarreferens i stället. De flesta scenarier använder standardprofilen som tillhandahålls av Azure IoT Operations.

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

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

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

Begära diskpersistence

Begärandediskens beständighet gör att dataflödesdiagram kan underhålla tillstånd över omstarter. När du aktiverar den här funktionen kan diagrammet återställa bearbetningstillståndet om den anslutna koordinatorn startas om. Den här funktionen är användbar för tillståndskänsliga bearbetningsscenarier där det skulle vara problematiskt att förlora mellanliggande data. När du aktiverar begärandediskpersistence bevarar mäklaren MQTT-data, till exempel meddelanden i prenumerantkön, till disken. Den här metoden säkerställer att dataflödets datakälla inte drabbas av dataförlust vid strömavbrott eller omstarter av asynkrona koordinatorer. Mäklaren upprätthåller optimala prestanda eftersom beständighet har konfigurerats per dataflöde, så endast de dataflöden som behöver beständighet använder den här funktionen.

Dataflödesdiagrammet gör denna beständighetsbegäran under prenumerationen med hjälp av en MQTTv5-användaregenskap. Den här funktionen fungerar bara när:

• Dataflödet använder MQTT-koordinatorn som källa (källnod med MQTT-slutpunkt)
• MQTT-koordinatorn har beständighet aktiverat med dynamiskt beständighetsläge inställt på för datatypen, till Enabled exempel prenumerantköer

Med den här konfigurationen kan MQTT-klienter som dataflödesdiagram begära diskpersistence för sina prenumerationer med hjälp av MQTTv5-användaregenskaper. Detaljerad beständighetskonfiguration för MQTT-koordinator finns i Konfigurera beständighet för MQTT-koordinator.

Inställningen accepterar Enabled eller Disabled, med Disabled som standard.

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

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

Nodkonfiguration

Noder är byggstenarna i ett dataflödesdiagram. Varje nod har ett unikt namn i diagrammet och utför en specifik funktion. Det finns tre typer av noder:

Källnoder

Källnoder definierar var data kommer in i diagrammet. De ansluter till dataflödesslutpunkter som tar emot data från MQTT-koordinatorer eller Kafka-ämnen. Varje källnod måste ange:

• Slutpunktsreferens som pekar på en konfigurerad dataflödesslutpunkt
• Datakällor som en lista över MQTT-ämnen eller Kafka-ämnen att prenumerera på
• Tillgångsreferens (valfritt) som länkar till en Azure Device Registry-tillgång för schemainferens

Med matrisen datakällor kan du prenumerera på flera ämnen utan att ändra slutpunktskonfigurationen. Den här flexibiliteten möjliggör återanvändning av slutpunkter i olika dataflöden.

{
  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

Diagrambearbetningsnoder

Diagrambearbetningsnoder innehåller WebAssembly-modulerna som transformerar data. Dessa noder hämtar WASM-artefakter från containerregister och kör dem med angivna konfigurationsparametrar. Varje grafnod kräver:

• Registerslutpunktsreferens som pekar på en registerslutpunkt för att hämta artefakter
• Artefaktspecifikation som definierar modulnamnet och versionen som ska hämtas
• Konfigurationsparametrar som nyckel/värde-par som skickas till WASM-modulen

Med konfigurationsmatrisen kan du anpassa modulbeteendet utan att återskapa WASM-artefakten. Vanliga konfigurationsalternativ är bearbetningsparametrar, tröskelvärden, konverteringsinställningar och funktionsflaggor.

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

Konfigurationens nyckel/värde-par skickas till WASM-modulen vid körning. Modulen kan komma åt dessa värden för att anpassa dess beteende. Med den här metoden kan du:

• Distribuera samma WASM-modul med olika konfigurationer
• Justera bearbetningsparametrar utan att återskapa moduler
• Aktivera eller inaktivera funktioner baserat på distributionskrav
• Ange miljöspecifika värden som tröskelvärden eller slutpunkter

Målnoder

Målnoder definierar var bearbetade data skickas. De ansluter till dataflödesslutpunkter som skickar data till MQTT-koordinatorer, molnlagring eller andra system. Varje målnod anger:

• Slutpunktsreferens som pekar på en konfigurerad dataflödesslutpunkt
• Datamål som det specifika ämnet, sökvägen eller platsen för utdata
• Inställningar för utdataschema (valfritt) som definierar serialiseringsformat och schemavalidering

För lagringsmål som Azure Data Lake eller Fabric OneLake kan du ange inställningar för utdataschema för att styra hur data serialiseras och verifieras.

{
  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

Nodanslutningar

Nodanslutningar definierar dataflödessökvägen mellan noder. Varje anslutning anger en källnod och en målnod som skapar bearbetningspipelinen. Anslutningar kan också inkludera schemavalidering för att säkerställa dataintegriteten mellan bearbetningsstegen.

När du anger schemaverifiering validerar systemet dataformatet och strukturen när det flödar mellan noder. Valideringen hjälper till att fånga in datainkonsekvenser tidigt och ser till att WASM-moduler tar emot data i förväntat format.

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

Slutpunkter för dataflöde

Dataflödesdiagram ansluter till externa system via dataflödesslutpunkter. Typen av slutpunkt avgör om den kan användas som källa, mål eller både och:

MQTT-slutpunkter

MQTT-slutpunkter kan fungera som både källor och mål. De ansluter till MQTT-koordinatorer, inklusive:

• Lokal MQTT-asynkron Azure IoT Operations-asynkron ( krävs i varje dataflöde)
• Azure Event Grid MQTT
• Anpassade MQTT-koordinatorer

Detaljerad konfigurationsinformation finns i Konfigurera MQTT-dataflödesslutpunkter.

Kafka-slutpunkter

Kafka-slutpunkter kan fungera som både källor och mål. De ansluter till Kafka-kompatibla system, inklusive:

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

Detaljerad konfigurationsinformation finns i Konfigurera Azure Event Hubs- och Kafka-dataflödesslutpunkter.

Lagringsslutpunkter

Lagringsslutpunkter kan bara fungera som mål. De ansluter till molnlagringssystem för långsiktig datakvarhållning och analys:

• Azure Data Lake Storage
• Microsoft Fabric OneLake
• Lokal lagring

Lagringsslutpunkter kräver vanligtvis schemainställningar för utdata för att definiera dataserialiseringsformat.

Registerslutpunkter

Registerslutpunkter ger åtkomst till containerregister för att hämta WASM-moduler och diagramdefinitioner. De används inte direkt i dataflödet, men diagrambearbetningsnoder refererar till dem.

Detaljerad konfigurationsinformation finns i Konfigurera registerslutpunkter.

Relaterat innehåll

• Konfigurera WebAssembly-grafdefinitioner
• Utveckla WebAssembly-moduler för dataflödesdiagram
• Konfigurera registerslutpunkter
• Konfigurera MQTT-slutpunkter för dataflöde
• Konfigurera Azure Event Hubs- och Kafka-slutpunkter för dataflöde
• Konfigurera Azure Data Lake Storage-slutpunkter för dataflöde
• Konfigurera Microsoft Fabric OneLake-dataflödesslutpunkter