Usa le trasformazioni WebAssembly (WASM) nei grafici di flusso dati

Operazioni di Azure IoT grafici dei flussi di dati includono trasformazioni predefinite per attività di elaborazione comuni, ad esempio mapping, filtro e aggregazione. Quando è necessaria una logica personalizzata oltre a quanto fornito dalle trasformazioni predefinite, è possibile distribuire moduli WebAssembly (WASM) come trasformazioni personalizzate nelle pipeline dei grafici del flusso di dati.

Prerequisiti

• Istanza di Operazioni di Azure IoT distribuita in un cluster Kubernetes. Per altre informazioni, vedere Deploy Operazioni di Azure IoT.

• Configurare un endpoint del registro per accedere ai moduli WASM e alle definizioni dei grafi. Sono disponibili due opzioni:
  • Avvio rapido con il Registro di sistema pubblico: creare un endpoint del Registro di sistema per ghcr.io con l'autenticazione anonima. Per istruzioni, vedere Usare moduli predefiniti da un registro pubblico.
  • Registro privato: configurare il proprio registro contenitori ed eseguire il push dei moduli di esempio seguendo le indicazioni riportate in Distribuire moduli WebAssembly (WASM) e definizioni di grafo.

Gli esempi interfaccia della riga di comando di Azure in questo articolo usano variabili di ambiente così puoi impostare ogni valore una volta e poi copiare e incollare i comandi as-is. Se stai usando l'ambiente Operazioni di Azure IoT Codespaces dal quickstart, queste variabili sono già impostate per te e puoi saltare questo passaggio. Altrimenti, imposta le seguenti variabili di ambiente nella tua shell prima di eseguire i comandi.

I seguenti script impostano le variabili di ambiente più comunemente utilizzate:

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

Devi solo impostare le variabili utilizzate in questo articolo. Questo articolo potrebbe utilizzare variabili ambientali aggiuntive per i nomi delle risorse che scegli. L'articolo spiega come posizionarli dove vengono introdotti.

Questo articolo utilizza anche le seguenti variabili ambientali per i valori che scegli: REGISTRY_ENDPOINT (il nome dell'endpoint del registro), GRAPH_NAME (il nome del grafo del flusso dati), PROFILE (il nome del profilo del flusso dati) e REGISTRY_HOST (il nome host del registro container). Imposta ognuno prima di eseguire i comandi correlati.

Informazioni generali

Usando moduli WebAssembly (WASM) nei grafici di flusso di dati di Operazioni di Azure IoT, è possibile elaborare i dati al perimetro con prestazioni e sicurezza elevate. WASM viene eseguito in un ambiente in modalità sandbox e supporta Rust e Python.

Un flusso di dati è una pipeline che sposta e trasforma i dati tra endpoint utilizzando trasformazioni integrate. Un grafico del flusso di dati estende i flussi di dati con passaggi di elaborazione componibili. Operazioni di Azure IoT fornisce grafici del flusso di dati predefiniti per operazioni comuni come mapping, filtro, diramazione e aggregazione. Per la logica di elaborazione personalizzata, implementa moduli WebAssembly come descritto in questo articolo. I grafici del flusso di dati usano definizioni di grafo YAML che specificano la modalità di connessione degli operatori. La risorsa del grafico del flusso di dati incapsula questa definizione e mappa le sue operazioni di origine astratta e sink a endpoint concreti, come i topic MQTT e i topic Kafka.

Funzionamento dei grafici del flusso di dati WASM

L'implementazione del flusso di dati WASM segue questo flusso di lavoro:

1. Sviluppare moduli WASM: scrivere la logica di elaborazione personalizzata in un linguaggio supportato e compilarla nel formato Modello componente WebAssembly. Per saperne di più, consulta Creare moduli WASM per i flussi di dati.
2. Sviluppare la definizione dei grafici: definire il modo in cui i dati si spostano tra i moduli usando i file di configurazione YAML. Per altre informazioni, vedere Configurare le definizioni dei grafici WebAssembly.
3. Archiviare gli artefatti nel Registro di sistema: eseguire il push dei moduli WASM compilati e delle definizioni del grafo in un registro contenitori usando strumenti compatibili con OCI, ad esempio ORAS. Per altre informazioni, vedere Distribuire moduli WebAssembly (WASM) e definizioni di grafo.
4. Configurare gli endpoint del Registro di sistema: configurare i dettagli di autenticazione e connessione in modo che Operazioni di Azure IoT possano accedere al registro contenitori. Per altre informazioni, vedere Configurare gli endpoint del Registro di sistema.
5. Creare un grafico del flusso di dati: usare l'interfaccia utente Web dell'esperienza operativa o i file Bicep per definire un flusso di dati che usa una definizione del grafo.
6. Distribuisci ed esegui: Operazioni di Azure IoT recupera le definizioni di grafici e i moduli WASM dal registro contenitori e li esegue.

Gli esempi seguenti illustrano come configurare grafici del flusso di dati WASM per scenari comuni. Gli esempi usano valori hardcoded e configurazioni semplificate per consentire di iniziare rapidamente.

Esempio 1: Distribuzione di base con un modulo WASM

Questo esempio converte i dati relativi alla temperatura da Fahrenheit a Celsius usando un modulo WASM. Il codice sorgente del modulo temperature è disponibile in GitHub. Se sono stati seguiti i passaggi di esempio in Distribuire moduli WebAssembly (WASM) e definizioni di grafo, la definizione del grafo e il graph-simple:1.0.0 modulo precompilato temperature:1.0.0 sono già presenti nel registro contenitori. Il percorso dell'artefatto del grafo di esempio è azure-samples/explore-iot-operations/graph-simple:1.0.0. Usare questo percorso per i campioni pubblici di GHCR o quando copi gli artefatti di esempio nel tuo registry con lo stesso percorso del repository.

Funzionamento del grafico semplice

La definizione graph crea una pipeline semplice a tre fasi:

1. Origine: riceve i dati sulla temperatura da MQTT
2. Mappa: elabora i dati con il modulo WASM TEMPERATURE
3. Sink: invia nuovamente i dati convertiti a MQTT

Per altre informazioni sul funzionamento della definizione del grafo semplice e sulla relativa struttura, vedere Esempio 1: Definizione di grafico semplice.

Formato di input:

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

Formato output:

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

La configurazione seguente crea un grafico del flusso di dati che usa questa pipeline di conversione della temperatura. Il grafico del flusso di dati fa riferimento alla definizione del grafo YAML graph-simple:1.0.0 e scarica il modulo di temperatura dal registro contenitori. Il grafo di esempio di Azure è archiviato nel percorso del repository azure-samples/explore-iot-operations, quindi includi tale percorso nel valore artifact.

Configurare il grafico del flusso di dati

Questa configurazione definisce tre nodi che implementano il flusso di lavoro di conversione della temperatura: un nodo di origine che sottoscrive i dati relativi alla temperatura in ingresso, un nodo di elaborazione del grafo che esegue il modulo WASM e un nodo di destinazione che pubblica i risultati convertiti.

La risorsa del grafico del flusso di dati incapsula l'artefatto della definizione del grafico e collega le operazioni astratte di origine e sink agli endpoint concreti.

• L'operazione source della definizione del grafico si connette al nodo di origine del flusso di dati (argomento MQTT)
• L'operazione sink della definizione del grafico si connette al nodo di destinazione del flusso di dati (argomento MQTT)
• Le operazioni di elaborazione della definizione del grafico vengono eseguite all'interno del nodo di elaborazione del grafico

Questa separazione consente di distribuire la stessa definizione del grafico con endpoint diversi nei diversi ambienti mantenendo al contempo invariata la logica di elaborazione.

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

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

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

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

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

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

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

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

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

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

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

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

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

kubectl apply -f dataflow-graph.yaml

Testare il flusso di dati

Per testare il flusso di dati, inviare messaggi MQTT dall'interno del cluster. Il modulo temperature prevede messaggi in un formato JSON specifico con un oggetto annidato temperature contenente value campi (numerici) e unit (stringa). Ad esempio: {"temperature":{"value":72,"unit":"F"}}.

Distribuire prima di tutto il pod client MQTT seguendo le istruzioni riportate in Testare la connettività al broker MQTT con i client MQTT. Il client MQTT fornisce i token di autenticazione e i certificati per la connessione al broker. Per distribuire il client MQTT, eseguire il comando seguente:

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

Inviare messaggi relativi alla temperatura

Nella prima sessione del terminale creare ed eseguire uno script per inviare dati sulla temperatura in Fahrenheit:

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

  echo "Publishing temperature: $payload"

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

  sleep 1
done'

Lo script pubblica dati di temperatura casuali nell'argomento sensor/temperature/raw ogni secondo. L'aspetto dovrebbe risultare simile al seguente:

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

Lasciare lo script in esecuzione per continuare a pubblicare i dati relativi alla temperatura.

Sottoscrivere messaggi elaborati

Nella seconda sessione del terminale, anch'essa connessa al pod client MQTT, sottoscrivere l'argomento di output per visualizzare i valori di temperatura convertiti:

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

Vedi che il modulo WASM converte i dati di temperatura da Fahrenheit a 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}}

Esempio 2: Distribuire un grafico complesso

In questo esempio viene illustrato un flusso di lavoro di elaborazione dati sofisticato che gestisce più tipi di dati, tra cui temperatura, umidità e dati di immagine. La definizione del grafo complex orchestra più moduli WASM per eseguire analisi avanzate e rilevamento degli oggetti.

Funzionamento del grafico complesso

Il grafico complesso elabora tre flussi di dati e li combina nell'analisi dei sensori arricchita:

• Elaborazione della temperatura: converte i dati da Fahrenheit in Celsius, filtra le letture non valide e calcola le statistiche
• Elaborazione dell'umidità: accumula le misurazioni dell'umidità negli intervalli di tempo
• Elaborazione immagini: esegue il rilevamento oggetti sugli snapshot della fotocamera e formatta i risultati

Per altre informazioni sul funzionamento della definizione complessa del grafo, sulla relativa struttura e sul flusso di dati attraverso più fasi di elaborazione, vedere Esempio 2: Definizione di grafico complesso.

Il grafico usa moduli specializzati della raccolta di operatori Rust.

Configurare il grafico del flusso di dati complesso

Questa configurazione implementa il workflow di elaborazione multisensore usando la definizione del graph-complex:1.0.0 grafo YAML. Il percorso dell'artefatto del grafo di esempio è azure-samples/explore-iot-operations/graph-complex:1.0.0. Si noti che la distribuzione del grafico del flusso di dati è simile all'esempio 1 : entrambi usano lo stesso modello a tre nodi (origine, processore grafico, destinazione) anche se la logica di elaborazione è diversa.

Questa somiglianza è presente perché la risorsa del grafico del flusso di dati funge da ambiente host che carica ed esegue definizioni di grafico. La logica di elaborazione effettiva risiede nella definizione del grafo (graph-simple:1.0.0 o graph-complex:1.0.0), che contiene la specifica YAML delle operazioni e delle connessioni tra i moduli WASM. La risorsa del grafo del flusso di dati fornisce l'infrastruttura di runtime per recuperare la definizione del grafo, istanziare i moduli e instradare i dati attraverso il flusso di lavoro definito.

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

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

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

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

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

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

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

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

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

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

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

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

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

Testare il flusso di dati complesso

Prima di visualizzare qualsiasi output, configurare i dati di origine.

Caricare file di immagine RAW nel pod mqtt-client

I file di immagine sono destinati al modulo snapshot per rilevare gli oggetti nelle immagini. I file sono nella cartella immagini su GitHub.

Prima di tutto, clonare il repository per ottenere l'accesso ai file di immagine:

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

Per caricare i file di immagine RAW dalla ./samples/wasm/images cartella al mqtt-client pod, usare il comando seguente:

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

Verificare che i file siano stati caricati:

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

Verrà visualizzato l'elenco dei file nella cartella /tmp/images.

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

Pubblicare dati di temperatura e umidità simulati e inviare immagini

È possibile combinare i comandi per la pubblicazione di dati relativi a temperatura e umidità e l'invio di immagini in un unico script. Usare il comando seguente:

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

Controllare l'output

In un nuovo terminale sottoscrivere l'argomento di output:

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

L'output ha un aspetto simile all'esempio seguente:

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

In questo caso, l'output contiene i dati relativi a temperatura e umidità, nonché gli oggetti rilevati nelle immagini.

Configurazione dei grafici del flusso di dati personalizzati

In questa sezione vengono fornite informazioni dettagliate sulla configurazione dei grafici del flusso di dati con i moduli WASM. Vengono illustrate tutte le opzioni di configurazione, gli endpoint del flusso di dati e le impostazioni avanzate.

Panoramica del grafico del flusso di dati

Un grafico del flusso di dati definisce il modo in cui i dati passano attraverso i moduli WebAssembly per l'elaborazione. Ogni grafico è costituito da:

• Modalità che controlla se il grafico è abilitato o disabilitato
• Riferimento al profilo che collega a un profilo del flusso di dati che definisce le impostazioni di ridimensionamento e risorse
• Persistenza del disco che, facoltativamente, abilita l'archiviazione permanente per lo stato del grafico
• Nodi che definiscono i componenti di origine, elaborazione e destinazione
• Connessioni di nodi che specificano il modo in cui i dati scorrono tra i nodi

Configurazione della modalità

La proprietà mode determina se il grafico del flusso di dati elabora attivamente i dati. Impostare la modalità su Enabled o Disabled (senza distinzione di maiuscole). Se è disabilitato, il grafico interrompe l'elaborazione dei dati ma ne mantiene la configurazione.

{
  "mode": "Enabled"
}

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

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

Riferimento al profilo

Il riferimento al profilo connette il grafico del flusso di dati a un profilo del flusso di dati, che definisce le impostazioni di ridimensionamento, i conteggi delle istanze e i limiti delle risorse. Se non si specifica un riferimento al profilo, è necessario usare invece un riferimento al proprietario di Kubernetes. La maggior parte degli scenari utilizza il profilo predefinito fornito da Operazioni di Azure IoT.

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

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

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

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

Richiedere la persistenza del disco

La persistenza su disco aiuta i grafi di flusso di dati a mantenere lo stato durante i riavvii. Quando attivi questa funzione, il grafo può recuperare lo stato di elaborazione se il broker connesso si riavvia. Questa funzionalità è utile per gli scenari di elaborazione con stato in cui la perdita di dati intermedi sarebbe problematica. Quando si abilita la richiesta di persistenza del disco, il broker salva in modo permanente i dati MQTT, ad esempio i messaggi nella coda del sottoscrittore, su disco. Questo approccio garantisce che l'origine dati del flusso di dati non subisca perdite di dati durante interruzioni dell'alimentazione o riavvii del broker. Il broker mantiene prestazioni ottimali perché si configura la persistenza per flusso di dati, quindi solo i flussi di dati che necessitano di persistenza utilizzano questa funzione.

Il grafico del flusso dati effettua questa richiesta di persistenza durante l'abbonamento utilizzando una proprietà utente MQTTv5. Questa funzionalità funziona solo quando:

• Il flusso di dati usa il broker MQTT come origine (nodo di origine con endpoint MQTT)
• Il broker MQTT ha la persistenza abilitata con la modalità di persistenza dinamica impostata su Enabled per il tipo di dati, ad esempio le code dei sottoscrittori

Con questa configurazione, i client MQTT come i grafici di flusso dati possono richiedere la persistenza del disco per i loro abbonamenti utilizzando le proprietà utente MQTTv5. Per informazioni dettagliate sulla configurazione della persistenza del broker MQTT, vedere Configurare la persistenza del broker MQTT.

L'impostazione accetta Enabled o Disabled, con Disabled come impostazione predefinita.

{
  "requestDiskPersistence": "Enabled"
}

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

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

Regole e limiti di denominazione

Le risorse del grafico del flusso di dati e i relativi componenti hanno vincoli di denominazione applicati a livelli diversi:

Il grafico del flusso di dati applica anche le regole strutturali seguenti:

• Nessun nome di nodo duplicato: ogni nodo nel grafico deve avere un nome univoco.
• Tipi di connessione validi: Il grafo consente solo i seguenti tipi di connessione di nodo: Sorgente a Grafo, Sorgente a Destinazione, Grafo a Grafo e Grafo a Destinazione.
• Nessun ciclo: il grafico non può contenere connessioni circolari che creerebbero cicli di elaborazione infiniti.
• Nessun ciclo automatico: un nodo non può connettersi a se stesso.
• Nessuna sovrapposizione di argomenti: se un'origine e una destinazione usano lo stesso endpoint, gli argomenti MQTT non possono sovrapporsi, in modo da creare un ciclo di messaggi infinito.

Configurazione nodi

I nodi sono i blocchi predefiniti di un grafico del flusso di dati. Ogni nodo ha un nome univoco all'interno del grafico ed esegue una funzione specifica. Il grafico include tre tipi di nodi:

Nodi di origine

I nodi di origine definiscono il punto in cui i dati entrano nel grafico. Si connettono agli endpoint del flusso di dati che ricevono dati da broker MQTT o topic Kafka. Ogni nodo di origine deve specificare:

• Riferimento all'endpoint che punta a un endpoint del flusso di dati configurato.
• Origini dati come elenco di argomenti MQTT o argomenti Kafka da sottoscrivere
• Riferimento asset (facoltativo) collegato a un asset del Registro Dispositivi Azure per l'inferenza dello schema

L'array delle origini dati supporta l'iscrizione a più argomenti senza modificare la configurazione dell'endpoint. Questa flessibilità significa che puoi riutilizzare endpoint tra diversi flussi di dati.

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

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

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

Nodi di elaborazione del grafico

I nodi di elaborazione del grafico contengono i moduli WebAssembly che trasformano i dati. Questi nodi eseguono il pull degli artefatti WASM dai registri contenitori e li eseguono usando i parametri di configurazione specificati. Ogni nodo del grafico richiede:

• Riferimento all'endpoint del registro che punta a un endpoint del registro per il pull degli artefatti
• Specifica dell'artefatto che definisce il nome del modulo e la versione di cui eseguire il pull
• Parametri di configurazione come coppie chiave-valore passate al modulo WASM

L'array di configurazione supporta la personalizzazione del comportamento dei moduli senza ricostruire l'artefatto WASM. Le opzioni di configurazione comuni includono parametri di elaborazione, soglie, impostazioni di conversione e flag di funzionalità.

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

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

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

Passi le coppie chiave-valore di configurazione al modulo WASM durante il runtime. Il modulo può accedere a questi valori per personalizzarne il comportamento. Utilizzando questo approccio, puoi:

• Distribuire lo stesso modulo WASM con configurazioni diverse.
• Modificare i parametri di elaborazione senza ricompilare i moduli.
• Abilitare o disabilitare le funzionalità in base ai requisiti di distribuzione.
• Impostare valori specifici dell'ambiente, ad esempio soglie o endpoint.

Nodi di destinazione

I nodi di destinazione definiscono dove il grafo invia i dati elaborati. Si connettono agli endpoint del flusso di dati che inviano dati a broker MQTT o ad altri sistemi. Ogni nodo di destinazione specifica:

• Riferimento all'endpoint che punta a un endpoint del flusso di dati configurato.
• Destinazione dati come elemento, posizione o percorso specifico per i dati di output.
• Impostazioni dello schema di output (facoltativo) che definiscono il formato di serializzazione e la convalida dello schema.

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

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

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

Connessioni ai nodi

Le connessioni ai nodi definiscono il percorso del flusso di dati tra i nodi. Ogni connessione specifica un nodo di origine e un nodo di destinazione, creando la pipeline di elaborazione. Facoltativamente, è possibile includere uno schema nella connessione. Il modulo riceve lo schema all'inizializzazione, che supporta la validazione dello schema come in questo esempio.

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

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

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

Endpoint del flusso di dati

I grafici del flusso di dati si connettono a sistemi esterni tramite endpoint del flusso di dati. Il tipo di endpoint determina se è possibile usarlo come origine, destinazione o entrambi.

Endpoint MQTT

Gli endpoint MQTT possono fungere sia da origini che da destinazioni. Si connettono a broker MQTT, tra cui:

• Operazioni di Azure IoT MQTT broker locale (obbligatorio in ogni flusso di dati)
• Griglia di eventi di Azure MQTT
• Broker MQTT personalizzati

Per informazioni dettagliate sulla configurazione, vedere Configurare gli endpoint del flusso di dati MQTT.

Endpoint Kafka

Gli endpoint Kafka possono fungere sia da origini che da destinazioni. Si connettono a sistemi compatibili con Kafka, tra cui:

• Hub eventi di Azure (compatibile con Kafka)
• Cluster Apache Kafka
• Nube Confluente

Per informazioni dettagliate sulla configurazione, vedere Configurare gli endpoint del flusso di dati Hub eventi di Azure e Kafka.

Endpoint del registro

Gli endpoint del registro forniscono l'accesso ai registri contenitori per il pull di moduli WASM e definizioni di grafo. Non vengono usati direttamente nel flusso di dati, ma i nodi di elaborazione del grafo vi fanno riferimento.

Per informazioni dettagliate sulla configurazione, vedere Configurare gli endpoint del registro.

Risolvere i problemi relativi ai grafici del flusso di dati

In questa sezione vengono forniti suggerimenti per la risoluzione dei problemi comuni relativi all'uso dei grafici del flusso di dati.

RegistryEndpoint non trovato

Se l'avvio del grafo del flusso di dati ha esito negativo e segnala che non riesce a trovare l'endpoint del registro, verificare quanto segue:

1. Corrisponde al nome dell'endpoint del registro: il valore registryEndpointRef nel grafico del flusso di dati deve corrispondere esattamente a quello name della risorsa RegistryEndpoint. Verificare la presenza di errori di battitura e la distinzione tra lettere maiuscole e minuscole.

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

2. L'endpoint del Registro di sistema si trova nello spazio dei nomi corretto: l'endpoint del Registro di sistema deve trovarsi nello azure-iot-operations spazio dei nomi o nello stesso spazio dei nomi del grafico del flusso di dati.

3. Endpoint del registro pronto: controllare lo stato dell'endpoint del registro.

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

4. Autenticazione è configurata correttamente: se utilizzi l'identità gestita, verifica che l'estensione Operazioni di Azure IoT Arc abbia le autorizzazioni AcrPull nel repository. Se si usa l'autenticazione anonima con un registro pubblico, verificare che l'URL host sia corretto.

5. Gli artefatti sono presenti nel Registro di sistema: verificare che la definizione del grafo e i moduli WASM a cui si fa riferimento nel grafico siano disponibili nei tag previsti nel Registro di sistema:

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

Il grafico del flusso di dati è in esecuzione ma non elabora i dati

Se distribuisci il grafico del flusso di dati ma non elabora i messaggi:

1. Controllare lo stato del grafico del flusso di dati: cercare gli errori nello stato della risorsa del grafico del flusso di dati.

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

2. Verificare gli argomenti MQTT: verificare che gli argomenti di origine nel grafico del flusso di dati corrispondano agli argomenti in cui si pubblicano i dati.

3. Controllare i timestamp: i grafici del flusso di dati usano timestamp HLC (Hybrid Logical Clock) per l'elaborazione dei messaggi. Includi la __ts proprietà utente quando pubblichi i messaggi MQTT per garantire un'elaborazione tempestiva.

Contenuti correlati

• Compilare moduli WASM per i flussi di dati sia per i flussi di lavoro di sviluppo IDE che basati sull'interfaccia della riga di comando
• Configurare le definizioni del grafo WebAssembly per la struttura YAML del grafo e i parametri di configurazione
• Distribuire moduli WASM e definizioni di grafo per la configurazione del registro e la gestione degli artefatti
• Esegui l'inferenza ONNX in WASM per l'integrazione di modelli di machine learning
• Configurare gli endpoint del flusso di dati MQTT
• Configurare gli endpoint per i flussi di dati di Hub eventi di Azure e Kafka
• Configurare gli endpoint del flusso di dati Azure Data Lake Storage
• Configurare gli endpoint del flusso di dati Microsoft Fabric OneLake