Utiliser WebAssembly (WASM) avec des graphiques de flux de données

Les graphiques de flux de données Opérations Azure IoT prennent en charge les modules WebAssembly (WASM) pour le traitement des données personnalisées en périphérie. Vous pouvez déployer des transformations de données et de logique métier personnalisées dans le cadre de vos pipelines de flux de données.

Conditions préalables

• Déployez une instance Azure IoT Operations sur un cluster Kubernetes avec Arc. Pour obtenir plus d’informations, consultez Déployer Opérations Azure IoT.
• Configurez votre registre de conteneurs et ajoutez les exemples de définitions de graphe et les modules WASM en suivant les instructions fournies dans Déployer des modules WebAssembly (WASM) et des définitions de graphiques.

Aperçu

Les modules WebAssembly (WASM) dans les graphiques de flux de données Opérations Azure IoT vous permettent de traiter des données en périphérie avec une sécurité et une haute performance. WASM s’exécute dans un environnement de bac à sable et prend en charge Rust et Python.

Fonctionnement des graphiques de flux de données WASM

L’implémentation du flux de données WASM suit ce workflow :

1. Développer des modules WASM : écrivez une logique de traitement personnalisée dans un langage pris en charge et compilez-la au format du modèle de composant WebAssembly. Pour plus d'informations, consultez les rubriques suivantes :
   • Générer des modules WASM pour les flux de données dans VS Code
   • Développer des modules WebAssembly (WASM)
2. Développer une définition de graphique : définissez la façon dont les données passent par les modules en utilisant des fichiers de configuration YAML. Pour plus d’informations, consultez Configurer les définitions de graphique WebAssembly.
3. Stockez les artefacts dans le Registre : envoyez (push) les modules WASM compilés et les définitions de graphiques à un registre de conteneurs à l’aide d’outils compatibles OCI tels que ORAS. Pour plus d’informations, consultez Déployer des modules webAssembly (WASM) et des définitions de graphique.
4. Configurer des points de terminaison de registre : configurez les informations sur l’authentification et la connexion afin qu’Opérations Azure IoT puisse accéder au registre de conteneurs. Pour plus d’informations, consultez Configurer des points de terminaison de Registre.
5. Créer un graphique de flux de données : utilisez l’interface utilisateur web des opérations ou les fichiers Bicep pour définir un flux de données qui utilise une définition de graphique.
6. Déployez et exécutez : Azure IoT Operations extrait les définitions de graphiques et les modules WASM à partir du registre de conteneurs et les exécute.

Les exemples suivants montrent comment configurer des graphiques de flux de données WASM pour des scénarios courants. Les exemples utilisent des valeurs codées en dur et des configurations simplifiées pour vous permettre de commencer rapidement.

Exemple 1 : déploiement de base avec un module WASM

Cet exemple convertit les données de température de Fahrenheit en Celsius en tirant parti d’un module WASM. Le code source du module de température est disponible sur GitHub. Si vous avez suivi les exemples d’étapes dans déployer des modules WebAssembly (WASM) et des définitions de graphiques, la définition de graphique et le graph-simple:1.0.0 module précompilé temperature:1.0.0 se trouvent déjà dans votre registre de conteneurs.

Fonctionnement

La définition de graphique crée un pipeline simple en trois étapes :

1. Source : reçoit les données de température de MQTT
2. Carte : traite les données avec le module WASM de température
3. Récepteur : renvoie les données converties à MQTT

Pour en savoir plus sur le fonctionnement de la définition de graphique simple et sa structure, consultez l’exemple 1 : Définition de graphique simple.

Format d’entrée :

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

Format de sortie :

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

La configuration suivante crée un graphique de flux de données qui utilise ce pipeline de conversion de température. Le graphique de flux de données fait référence à la graph-simple:1.0.0 définition de graphique YAML et extrait le module de température de votre registre de conteneurs.

Configurer un graphique de flux de données

Cette configuration définit trois nœuds qui implémentent le workflow de conversion de la température : un nœud source qui s’abonne aux données de température entrantes, un nœud de traitement de graphique qui exécute le module WASM et un nœud de destination qui publie les résultats convertis.

La ressource graphique de flux de données encapsule l’artefact de définition du graphique et connecte ses opérations source/sink abstraites aux points de terminaison concrets :

• L’opération source de la définition de graphique se connecte au nœud source du flux de données (rubrique MQTT)
• L’opération sink de la définition de graphique se connecte au nœud de destination du flux de données (rubrique MQTT)
• Les opérations de traitement de la définition de graphique s’exécutent au sein du nœud de traitement du graphique

Cette séparation vous permet de déployer la même définition de graphique avec différents points de terminaison dans différents environnements tout en conservant la logique de traitement inchangée.

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

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

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

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

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

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

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

    - nodeType: Graph
      name: temperature-processor
      graphSettings:
        registryEndpointRef: <REGISTRY_ENDPOINT_NAME>
        artifact: graph-simple:1.0.0
        configuration:
          - key: key1
            value: example-value

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

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

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

kubectl apply -f dataflow-graph.yaml

Tester le flux de données

Pour tester le flux de données, envoyez des messages MQTT à partir du cluster. Tout d’abord, déployez le pod client MQTT en suivant les instructions dans Tester la connectivité à l’Agent MQTT avec des clients MQTT. Le client MQTT fournit les jetons d’authentification et les certificats pour se connecter au broker. Pour déployer le client MQTT, exécutez la commande suivante :

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

Envoyer des messages de température

Dans la première session de terminal, créez et exécutez un script pour envoyer des données de température en 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'

Le script publie des données de température aléatoires dans la rubrique sensor/temperature/raw toutes les secondes. Il doit se présenter comme suit :

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

Laissez le script s’exécuter pour continuer à publier les données de température.

S’abonner aux messages traités

Dans la deuxième session de terminal (également connectée au pod client MQTT), abonnez-vous à la rubrique de sortie pour afficher les valeurs de température converties :

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

Vous voyez les données de température converties de degrés Fahrenheit en Celsius par le module WASM.

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

Exemple 2 : déployer un graphique complexe

Cet exemple illustre un workflow de traitement de données sophistiqué qui gère plusieurs types de données, notamment la température, l’humidité et les données d’images. La définition de graphique complexe orchestre plusieurs modules WASM pour effectuer une analyse avancée et une détection d’objets.

Fonctionnement

Le graphique complexe traite trois flux de données et les combine en analytique de capteur enrichie :

• Traitement de la température : convertit de degrés Fahrenheit en Celsius, filtre les lectures non valides et calcule les statistiques
• Traitement de l’humidité : accumule les mesures d’humidité au fil des intervalles de temps
• Traitement d’images : effectue la détection d’objets sur les captures instantanées de caméra et met en forme les résultats

Pour en savoir plus sur le fonctionnement de la définition de graphique complexe, sa structure et le flux de données à travers plusieurs étapes de traitement, consultez l’exemple 2 : Définition de graphique complexe.

Le graphique utilise des modules spécialisés à partir de la collection d’opérateurs Rust.

Configurer le graphique de flux de données complexe

Cette configuration implémente le flux de travail de traitement multi-capteur à l’aide de la définition de graphique YAML graph-complex:1.0.0. Notez que le déploiement du graphe de flux de données est similaire à l’exemple 1 : les deux utilisent le même modèle à trois nœuds (source, processeur de graphe, destination) même si la logique de traitement est différente.

Cette similarité se produit, car la ressource de graphique de flux de données agit comme environnement hôte qui charge et exécute des définitions de graphiques. La logique de traitement réelle réside dans la définition de graphique (graph-simple:1.0.0 ou graph-complex:1.0.0), qui contient la spécification YAML des opérations et des connexions entre les modules WASM. La ressource de graphique de flux de données fournit l’infrastructure d’exécution pour extraire la définition de graphe, instancier les modules et acheminer les données via le flux de travail défini.

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

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

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

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

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

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

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

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

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

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

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

Tester le flux de données complexe

Avant de voir n’importe quelle sortie, configurez les données sources.

Charger des fichiers image RAW dans le pod mqtt-client

Les fichiers image sont destinés au module snapshot pour détecter des objets dans les images. Ils se trouvent dans le dossier images sur GitHub.

Tout d’abord, clonez le référentiel pour accéder aux fichiers image :

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

Pour charger des fichiers image RAW à partir du dossier ./samples/wasm/images vers le pod mqtt-client, vous pouvez utiliser la commande suivante :

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

Vérifiez que les fichiers sont chargés :

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

Vous devez voir la liste des fichiers dans le dossier /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

Publier des données de température simulée, d’humidité et envoyer des images

Vous pouvez combiner les commandes pour la publication des données de température, d’humidité et l’envoi d’images dans un seul script. Utilisez la commande suivante :

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

Consulter la sortie

Dans un nouveau terminal, abonnez-vous à la rubrique de sortie :

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

La sortie ressemble à l’exemple suivant :

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

Ici, la sortie contient les données de température et d’humidité, ainsi que les objets détectés dans les images.

Configuration des graphiques de flux de données personnalisés

Cette section fournit des informations détaillées sur la configuration des graphiques de flux de données avec des modules WASM. Elle couvre toutes les options de configuration, les points de terminaison de flux de données et les paramètres avancés.

Vue d’ensemble du graphique de flux de données

Un graphique de flux de données définit la façon dont les données circulent via des modules WebAssembly pour le traitement. Chaque graphique se compose des éléments suivants :

• Mode qui contrôle si le graphique est activé ou désactivé
• Référence de profil qui lie à un profil de flux de données définissant la mise à l’échelle et les paramètres de la ressource
• Persistance de disque qui active éventuellement le stockage persistant pour l’état d’un graphique
• Nœuds qui définissent les composants source, de traitement et de destination
• Connexions de nœud qui spécifient la façon dont les données circulent entre les nœuds

Configuration du mode

La propriété du mode détermine si le graphique de flux de données traite activement les données. Vous pouvez définir le mode Enabled sur ou Disabled (ne respecte pas la casse). Une fois désactivé, le graphique cesse de traiter les données, mais conserve sa configuration.

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

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

Informations de référence sur le profil

La référence de profil connecte votre graphique de flux de données à un profil de flux de données qui définit les paramètres de mise à l’échelle, le nombre d’instances et les limites de ressources. Si vous ne spécifiez pas de référence de profil, vous devez utiliser une référence de propriétaire Kubernetes à la place. La plupart des scénarios utilisent le profil par défaut fourni par Opérations Azure IoT.

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

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

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

Demander la persistance du disque

La persistance des disques de requête permet aux graphiques de flux de données de maintenir l’état entre les redémarrages. Lorsque vous activez cette fonctionnalité, le graphique peut récupérer l’état de traitement en cas de redémarrage du broker connecté. Cette fonctionnalité est utile pour les scénarios de traitement avec état où la perte de données intermédiaires sera problématique. Lorsque vous activez la persistance des disques de requête, le broker conserve les données MQTT, telles que les messages dans la file d’attente de l’abonné, sur le disque. Cette approche veille à ce que la source de données de votre flux de données ne subisse pas de perte de données pendant les pannes de courant ou les redémarrages du broker. Le broker conserve des performances optimales, car la persistance est configurée par flux de données. Par conséquent, seuls les flux de données qui ont besoin de persistance utilisent cette fonctionnalité.

Le graphique de flux de données effectue cette demande de persistance pendant l’abonnement en utilisant une propriété utilisateur MQTTv5. Cette fonctionnalité fonctionne uniquement lorsque :

• Le flux de données utilise l’Agent MQTT en tant que source (nœud source avec point de terminaison MQTT)
• L’Agent MQTT a la persistance activée avec le mode de persistance dynamique défini sur Enabled pour le type de données, comme les files d’attente des abonnés

Cette configuration permet aux clients MQTT tels que les graphiques de flux de données de demander la persistance de disque pour leurs abonnements en tirant parti des propriétés utilisateur MQTTv5. Pour découvrir la configuration détaillée de persistance du répartiteur MQTT, consultez Configurer la persistance de l’Agent MQTT.

Le paramètre accepte Enabled ou Disabled, avec Disabled comme valeur par défaut.

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

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

Configuration de nœuds

Les nœuds sont les blocs de construction d’un graphique de flux de données. Chaque nœud a un nom unique dans le graphique et exécute une fonction spécifique. Il existe trois types de nœuds :

Nœuds sources

Les nœuds sources définissent l’emplacement où les données entrent dans le graphique. Ils se connectent aux points de terminaison de flux de données qui reçoivent des données des répartiteurs MQTT ou des rubriques Kafka. Chaque nœud source doit spécifier :

• Référence de point de terminaison qui pointe vers un point de terminaison de flux de données configuré
• Sources de données sous forme de liste de rubriques MQTT ou de rubriques Kafka à laquelle s’abonner
• Référence de ressource (facultative) qui lie une ressource Registre de Dispositifs Azure pour l’inférence de schéma

Le tableau de sources de données vous permet de vous abonner à plusieurs rubriques sans modifier la configuration du point de terminaison. Cette flexibilité permet la réutilisation des points de terminaison sur différents flux de données.

{
  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

Nœuds de traitement de graphique

Les nœuds de traitement de graphique contiennent les modules WebAssembly qui transforment les données. Ces nœuds extraient des artefacts WASM à partir de registres de conteneurs et les exécutent avec des paramètres de configuration spécifiés. Chaque nœud de graphique nécessite :

• Référence de point de terminaison de registre qui pointe vers un point de terminaison de registre pour l’extraction d’artefacts
• Spécification d’artefact qui définit le nom et la version du module à extraire
• Paramètres de configuration en tant que paires clé-valeur transmises au module WASM

Le tableau de configuration vous permet de personnaliser le comportement du module sans reconstruire l’artefact WASM. Les options de configuration courantes incluent les paramètres de traitement, les seuils, les paramètres de conversion et les indicateurs de fonctionnalité.

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

Les paires clé-valeur de configuration sont transmises au module WASM au moment de l’exécution. Le module peut accéder à ces valeurs pour personnaliser son comportement. Cette approche vous permet de :

• Déployer le même module WASM avec différentes configurations
• Ajuster les paramètres de traitement sans reconstruire les modules
• Activer ou désactiver des fonctionnalités en fonction des exigences de déploiement
• Définir des valeurs spécifiques à l’environnement telles que des seuils ou des points de terminaison

Nœuds de destination

Les nœuds de destination définissent l’emplacement où les données traitées sont envoyées. Ils se connectent aux points de terminaison de flux de données qui envoient des données aux Agents MQTT, au stockage cloud ou à d’autres systèmes. Chaque nœud de destination spécifie :

• Référence de point de terminaison qui pointe vers un point de terminaison de flux de données configuré
• Destination des données en tant que rubrique, chemin d’accès ou emplacement spécifiques pour les données de sortie
• Paramètres de schéma de sortie (facultatif) qui définissent le format de sérialisation et la validation de schéma

Pour les destinations de stockage telles qu’Azure Data Lake ou Fabric OneLake, vous pouvez spécifier les paramètres de schéma de sortie pour contrôler la sérialisation et la validation des données.

{
  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

Connexions de nœud

Les connexions de nœud définissent le chemin du flux de données entre les nœuds. Chaque connexion spécifie un nœud source et un nœud de destination, créant le pipeline de traitement. Les connexions peuvent éventuellement inclure un schéma, fourni au module lors de l’initialisation, ce qui permet la validation de schéma comme dans cet exemple.

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

Points de terminaison de flux de données

Les graphiques de flux de données se connectent à des systèmes externes via des points de terminaison de flux de données. Le type de point de terminaison détermine s’il peut être utilisé comme source, destination ou les deux :

Points de terminaison MQTT

Les points de terminaison MQTT peuvent servir de sources et de destinations. Ils se connectent aux Agents MQTT, notamment :

• Agent MQTT local Opérations Azure IoT (requis dans chaque flux de données)
• Azure Event Grid MQTT
• Agents MQTT personnalisés

Pour obtenir des informations de configuration détaillées, consultez Configurer des points de terminaison de flux de données MQTT.

Points de terminaison Kafka

Les points de terminaison Kafka peuvent servir de sources et de destinations. Ils se connectent à des systèmes compatibles avec Kafka, notamment :

• Azure Event Hubs (compatible avec Kafka)
• Clusters Apache Kafka
• Confluent Cloud

Pour obtenir des informations de configuration détaillées, consultez Configurer des points de terminaison de flux de données Azure Event Hubs et Kafka.

Points de terminaison de stockage

Les points de terminaison de stockage peuvent uniquement servir de destinations. Ils se connectent aux systèmes de stockage cloud pour la rétention et l’analytique des données à long terme :

• Azure Data Lake Storage
• Microsoft Fabric OneLake
• Stockage local

Les points de terminaison de stockage nécessitent généralement des paramètres de schéma de sortie pour définir le format de sérialisation des données.

Points de terminaison de registre

Les points de terminaison de registre permettent d’accéder aux registres de conteneurs pour extraire des modules WASM et des définitions de graphiques. Ils ne sont pas utilisés directement dans le flux de données, mais les nœuds de traitement de graphique les référencent.

Pour obtenir des informations de configuration détaillées, consultez Configurer des points de terminaison de registre.

Contenu connexe

• Configurer des points de terminaison de flux de données MQTT
• Configurer des points de terminaison de flux de données Azure Event Hubs et Kafka
• Configurer des points de terminaison de flux de données Azure Data Lake Storage
• Configurer des points de terminaison de flux de données Microsoft Fabric OneLake