Configurer des flux de données dans Opérations Azure IoT

Important

Opérations Azure IoT Préversion avec Azure Arc est actuellement en préversion. Vous ne devez pas utiliser ce logiciel en préversion dans des environnements de production.

Lorsqu’une version en disponibilité générale sera publiée, vous devrez déployer une nouvelle installation d’Opérations Azure IoT. Vous ne pourrez pas mettre à niveau une installation en préversion.

Pour connaître les conditions juridiques qui s’appliquent aux fonctionnalités Azure en version bêta, en préversion ou plus généralement non encore en disponibilité générale, consultez l’Avenant aux conditions d’utilisation des préversions de Microsoft Azure.

Un flux de données est le chemin qu’empruntent les données de la source à la destination avec des transformations facultatives. Vous pouvez configurer le flux de données en créant une ressource personnalisée de Flux de données ou à l’aide du portail Studio d’Opérations Azure IoT. Un flux de données est constitué de trois parties : la source, la transformation et la destination.

Pour définir la source et la destination, vous devez configurer les points de terminaison de flux de données. La transformation est facultative et peut inclure des opérations telles que l’enrichissement des données, le filtrage des données et le mappage des données à un autre champ.

Important

Chaque flux de données doit avoir le point de terminaison par défaut de l’Agent MQTT Opérations Azure IoT local comme soit la source soit la destination.

Vous pouvez utiliser l’expérience des opérations dans Opérations Azure IoT pour créer un flux de données. L’expérience des opérations fournit une interface visuelle pour configurer le flux de données. Vous pouvez également utiliser Bicep pour créer un flux de données à l’aide d’un fichier de modèle Bicep ou utiliser Kubernetes pour créer un flux de données à l’aide d’un fichier YAML.

Poursuivez la lecture pour apprendre à configurer la source, la transformation et la destination.

Prérequis

Vous pouvez déployer des flux de données dès que vous disposez d’une instance de Préversion Opérations Azure IoT à l’aide du profil de flux de données et du point de terminaison par défaut. Toutefois, vous pouvez configurer des profils et des points de terminaison de flux de données pour personnaliser le flux de données.

Profil de flux de données

Le profil de flux de données spécifie le nombre d’instances pour les flux de données qu’il doit utiliser. Si vous n’avez pas besoin de plusieurs groupes de flux de données avec différents paramètres de mise à l’échelle, vous pouvez utiliser le profil de flux de données par défaut. Pour savoir comment configurer un profil de flux de données, consultez Configurer des profils de flux de données.

Points de terminaison de flux de données

Les points de terminaison du flux de données sont nécessaires pour configurer la source et la destination du flux de données. Pour commencer rapidement, vous pouvez utiliser le point de terminaison de flux de données par défaut pour le répartiteur MQTT local. Vous pouvez également créer d’autres types de points de terminaison de flux de données tels que Kafka, Event Hubs ou Azure Data Lake Storage. Pour savoir comment configurer chaque type de point de terminaison de flux de données, consultez Configurer des points de terminaison de flux de données.

Démarrage

Une fois que vous avez les conditions préalables, vous pouvez commencer à créer un flux de données.

Portail

Pour créer un flux de données dans l’expérience opérations, sélectionnez Flux de données>Créer un flux de données. Ensuite, vous voyez la page dans laquelle vous pouvez configurer la source, la transformation et la destination du flux de données.

Bicep

Créez un fichier bicep .bicep pour commencer à créer un flux de données. Cet exemple montre la structure du flux de données contenant les configurations source, de transformation et de destination.

param aioInstanceName string = '<AIO_INSTANCE_NAME>'
param customLocationName string = '<CUSTOM_LOCATION_NAME>'
param dataflowName string = '<DATAFLOW_NAME>'

resource aioInstance 'Microsoft.IoTOperations/instances@2024-09-15-preview' existing = {
  name: aioInstanceName
}

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

resource defaultDataflowEndpoint 'Microsoft.IoTOperations/instances/dataflowEndpoints@2024-09-15-preview' existing = {
  parent: aioInstance
  name: 'default'
}

// Pointer to the default dataflow profile
resource defaultDataflowProfile 'Microsoft.IoTOperations/instances/dataflowProfiles@2024-09-15-preview' existing = {
  parent: aioInstance
  name: 'default'
}

resource dataflow 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflows@2024-09-15-preview' = {
  // Reference to the parent dataflow profile, the default profile in this case
  // Same usage as profileRef in Kubernetes YAML
  parent: defaultDataflowProfile
  name: dataflowName
  extendedLocation: {
    name: customLocation.id
    type: 'CustomLocation'
  }
  properties: {
    mode: 'Enabled'
    operations: [
      {
        operationType: 'Source'
        sourceSettings: {
          // See source configuration section
        }
      }
      // Transformation optional
      {
        operationType: 'BuiltInTransformation'
        builtInTransformationSettings: {
          // See transformation configuration section
        }
      }
      {
        operationType: 'Destination'
        destinationSettings: {
          // See destination configuration section
        }
      }
    ]
  }
}

Kubernetes

Créez un manifeste Kubernetes .yaml fichier pour commencer à créer un flux de données. Cet exemple montre la structure du flux de données contenant les configurations source, de transformation et de destination.

apiVersion: connectivity.iotoperations.azure.com/v1beta1
kind: Dataflow
metadata:
  name: <DATAFLOW_NAME>
  namespace: azure-iot-operations
spec:
  # Reference to the default dataflow profile
  # This field is required when configuring via Kubernetes YAML
  # The syntax is different when using Bicep
  profileRef: default 
  mode: Enabled
  operations:
    - operationType: Source
      sourceSettings:
        # See source configuration section
      # Transformation optional
    - operationType: BuiltInTransformation
      builtInTransformationSettings:
        # See transformation configuration section
    - operationType: Destination
      destinationSettings:
        # See destination configuration section

Passez en revue les sections suivantes pour apprendre à configurer les types d’opérations du flux de données.

Source

Pour configurer une source pour le flux de données, spécifiez la référence du point de terminaison et une liste de sources de données pour le point de terminaison.

Utiliser la ressource comme source

Portail

Vous pouvez utiliser une ressource comme source du flux de données. L’utilisation d’un actif comme source n’est possible que dans le cadre de l’expérience opérationnelle.

1. Sous Détails de la source, sélectionnez Ressource.

2. Sélectionnez la ressource que vous souhaitez utiliser comme point de terminaison source.

3. Sélectionnez Continuer.

   Une liste de points de données pour la ressource sélectionnée s’affiche.

   

4. Sélectionnez Appliquer pour utiliser la ressource comme point de terminaison source.

Bicep

La configuration d’une ressource en tant que source n’est disponible que dans l’expérience des opérations.

Kubernetes

La configuration d’une ressource en tant que source n’est disponible que dans l’expérience des opérations.

Lors de l’utilisation d’une ressource comme source, la définition de ressource est utilisée pour déduire le schéma du flux de données. La définition de ressource inclut le schéma des points de données de la ressource. Pour en savoir plus, consultez Gérer les configurations de ressources à distance.

Une fois configurées, les données de la ressource ont atteint le flux de données via l’Agent MQTT local. Par conséquent, lors de l’utilisation d’une ressource comme source, le flux de données utilise le point de terminaison par défaut de l’Agent MQTT local comme source en réalité.

Utiliser le point de terminaison MQTT par défaut comme source

Portail

1. Sous Détails de la source, sélectionnez MQTT.

   

2. Saisissez les paramètres suivants pour la source MQTT :

   Setting	Description
   Rubrique MQTT	Filtre de rubrique MQTT auquel s’abonner pour les messages entrants. Voir Configurer les rubriques MQTT ou Kafka.
   Schéma du message	Le schéma à utiliser pour désérialiser les messages entrants. Voir Spécifier le schéma pour désérialiser les données.

3. Sélectionnez Appliquer.

Bicep

Le point de terminaison MQTT est configuré dans le fichier de modèle Bicep. Par exemple, le point de terminaison suivant est une source pour le flux de données.

sourceSettings: {
  endpointRef: 'default'
  dataSources: [
    'thermostats/+/telemetry/temperature/#'
    'humidifiers/+/telemetry/humidity/#'
  ]
}

Ici, dataSources vous permet de spécifier plusieurs rubriques MQTT ou Kafka sans avoir à modifier la configuration du point de terminaison. Cette flexibilité signifie que le même point de terminaison peut être réutilisé sur plusieurs flux de données, même si les rubriques varient. Pour en savoir plus, voir Configurer les sources de données.

Kubernetes

Par exemple, pour configurer une source à l’aide d’un point de terminaison MQTT et de deux filtres de rubrique MQTT, utilisez la configuration suivante :

sourceSettings:
  endpointRef: default
  dataSources:
    - thermostats/+/telemetry/temperature/#
    - humidifiers/+/telemetry/humidity/#

Étant donné que dataSources vous permet de spécifier des rubriques MQTT ou Kafka sans modifier la configuration du point de terminaison, vous pouvez réutiliser le point de terminaison pour plusieurs flux de données même si les rubriques sont différentes. Pour en savoir plus, voir Configurer les sources de données.

Si le point de terminaison par défaut n’est pas utilisé comme source, il doit être utilisé comme destination. Pour en savoir plus, consultez Dataflows doit utiliser le point de terminaison de l’Agent MQTT local.

Utiliser un point de terminaison de flux de données MQTT ou Kafka personnalisé comme source

Si vous avez créé un point de terminaison de flux de données MQTT ou Kafka personnalisé (par exemple, pour l’utiliser avec Event Grid ou Event Hubs), vous pouvez l’utiliser comme source pour le flux de données. N’oubliez pas que les points de terminaison de type de stockage, comme Data Lake ou Fabric OneLake, ne peuvent pas être utilisés comme source.

Pour configurer, utilisez Kubernetes YAML ou Bicep. Remplacez les valeurs d’espace réservé par le nom et les rubriques de votre point de terminaison personnalisé.

Portail

L’utilisation d’un point de terminaison MQTT ou Kafka personnalisé en tant que source n’est actuellement pas prise en charge dans l’expérience des opérations.

Bicep

sourceSettings: {
  endpointRef: '<CUSTOM_ENDPOINT_NAME>'
  dataSources: [
    '<TOPIC_1>'
    '<TOPIC_2>'
    // See section on configuring MQTT or Kafka topics for more information
  ]
}

Kubernetes

sourceSettings:
  endpointRef: <CUSTOM_ENDPOINT_NAME>
  dataSources:
  - <TOPIC_1>
  - <TOPIC_2>
  # See section on configuring MQTT or Kafka topics for more information

Configurer des sources de données (rubriques MQTT ou Kafka)

Vous pouvez spécifier plusieurs sujets MQTT ou Kafka dans une source sans avoir à modifier la configuration du point d’arrivée du flux de données. Cette flexibilité signifie que le même point de terminaison peut être réutilisé sur plusieurs flux de données, même si les rubriques varient. Pour plus d’informations, consultez Réutiliser les points de terminaison des flux de données.

Rubriques MQTT

Lorsque la source est un point de terminaison MQTT (Event Grid inclus), vous pouvez utiliser le filtre de rubrique MQTT pour vous abonner aux messages entrants. Le filtre de rubrique peut inclure des caractères génériques pour s’abonner à plusieurs rubriques. Par exemple, thermostats/+/telemetry/temperature/# s’abonne à tous les messages de télémétrie de température provenant de thermostats. Pour configurer les filtres de rubrique MQTT :

Portail

Dans le flux de données d’expérience des opérations détails de la source, sélectionnez MQTT, puis utilisez le champ rubrique MQTT pour spécifier le filtre de rubrique MQTT auquel s’abonner pour les messages entrants.

Remarque

Un seul filtre de rubrique MQTT peut être spécifié dans l’expérience des opérations. Pour utiliser plusieurs filtres de rubrique MQTT, utilisez Bicep ou Kubernetes.

Bicep

sourceSettings: {
  endpointRef: '<MQTT_ENDPOINT_NAME>'
  dataSources: [
    '<MQTT_TOPIC_FILTER_1>'
    '<MQTT_TOPIC_FILTER_2>'
    // Add more MQTT topic filters as needed
  ]
}

Exemple avec plusieurs filtres de rubrique MQTT avec des caractères génériques :

sourceSettings: {
  endpointRef: 'default'
  dataSources: [
    'thermostats/+/telemetry/temperature/#'
    'humidifiers/+/telemetry/humidity/#'
  ]
}

Ici, le + générique est utilisé pour sélectionner tous les appareils sous les rubriques thermostats et humidifiers. Le caractère générique # est utilisé pour sélectionner tous les messages de télémétrie sous toutes les sous-rubriques de temperature et de humidity.

Kubernetes

sourceSettings:
  endpointRef: <MQTT_ENDPOINT_NAME>
  dataSources:
    - <MQTT_TOPIC_FILTER_1>
    - <MQTT_TOPIC_FILTER_2>
    # Add more MQTT topic filters as needed

Exemple avec plusieurs filtres de rubrique MQTT avec des caractères génériques :

sourceSettings:
  endpointRef: default
  dataSources:
    - thermostats/+/telemetry/temperature/#
    - humidifiers/+/telemetry/humidity/#

Ici, le + générique est utilisé pour sélectionner tous les appareils sous les rubriques thermostats et humidifiers. Le caractère générique # est utilisé pour sélectionner tous les messages de télémétrie sous toutes les sous-rubriques de temperature et de humidity.

Abonnements partagés

Pour utiliser des abonnements partagés avec des sources MQTT, vous pouvez spécifier la rubrique d’abonnement partagé sous la forme de $shared/<GROUP_NAME>/<TOPIC_FILTER>.

Portail

Dans les détails de la source du flux de données de l’expérience opérationnelle, sélectionnez MQTT et utilisez le champ Rubrique MQTT pour spécifier le groupe d’abonnement partagé et le sujet.

Bicep

sourceSettings: {
  dataSources: [
    '$shared/<GROUP_NAME>/<TOPIC_FILTER>'
  ]
}

Kubernetes

sourceSettings:
  dataSources:
    - $shared/<GROUP_NAME>/<TOPIC_FILTER>

Si le nombre d’instances dans le profil de flux de données est supérieur à 1, l’abonnement partagé est automatiquement activé pour tous les flux de données qui utilisent la source MQTT. Dans ce cas, le préfixe $shared est ajouté et le nom du groupe d’abonnement partagé généré automatiquement. Par exemple, si vous avez un profil de flux de données avec un nombre d’instances de 3, et que votre flux de données utilise un point de terminaison MQTT comme source, configuré avec des rubriques topic1 et topic2, elles sont automatiquement converties en abonnements partagés en tant que $shared/<GENERATED_GROUP_NAME>/topic1 et $shared/<GENERATED_GROUP_NAME>/topic2. Si vous souhaitez utiliser un autre ID de groupe d’abonnement partagé, vous pouvez le remplacer dans la rubrique, par exemple $shared/mygroup/topic1.

Important

Les flux de données nécessitant un abonnement partagé lorsque le nombre d’instances est supérieur à 1 est important lors de l’utilisation de l’Agent MQTT Event Grid comme source, car il ne prend pas en charge les abonnements partagés. Pour éviter de manquer des messages, définissez le nombre d’instances de profil de flux de données sur 1 lors de l’utilisation de l’Agent MQTT Event Grid comme source. C’est-à-dire lorsque le flux de données est l’abonné et reçoit des messages depuis le cloud.

Rubriques Kafka

Lorsque la source est un point de terminaison Kafka (Event Hubs inclus), spécifiez les rubriques kafka individuelles auxquelles s’abonner pour les messages entrants. Les caractères génériques ne sont pas pris en charge. Vous devez donc spécifier chaque rubrique statiquement.

Remarque

Lorsque vous utilisez Event Hubs via le point de terminaison Kafka, chaque hub d’événements individuel au sein de l’espace de noms est la rubrique Kafka. Par exemple, si vous avez un espace de noms Event Hubs avec deux hubs d’événements, thermostats et humidifiers, vous pouvez spécifier chaque hub d’événements en tant que rubrique Kafka.

Pour configurer les thèmes Kafka :

Portail

L’utilisation d’un point de terminaison Kafka en tant que source n’est actuellement pas prise en charge dans l’expérience des opérations.

Bicep

sourceSettings: {
  endpointRef: '<KAFKA_ENDPOINT_NAME>'
  dataSources: [
    '<KAFKA_TOPIC_1>'
    '<KAFKA_TOPIC_2>'
    // Add more Kafka topics as needed
  ]
}

Kubernetes

sourceSettings:
  endpointRef: <KAFKA_ENDPOINT_NAME>
  dataSources:
    - <KAFKA_TOPIC_1>
    - <KAFKA_TOPIC_2>
    # Add more Kafka topics as needed

Spécifier le schéma pour désérialiser les données

Si les données sources ont des champs ou des champs facultatifs avec différents types, spécifiez un schéma de désérialisation pour garantir la cohérence. Par exemple, les données peuvent avoir des champs qui ne sont pas présents dans tous les messages. Sans le schéma, la transformation ne peut pas gérer ces champs, car ils auraient des valeurs vides. Avec le schéma, vous pouvez spécifier des valeurs par défaut ou ignorer les champs.

La spécification du schéma n’est pertinente que lors de l’utilisation de la source MQTT ou Kafka. Si la source est une ressource, le schéma est déduit automatiquement de la définition de ressource.

Pour configurer le schéma utilisé pour désérialiser les messages entrants à partir d’une source :

Portail

Dans le flux de données de l’expérience des opérations détails de la source, sélectionnez MQTT et utilisez le champ schéma de message pour spécifier le schéma. Vous pouvez utiliser le bouton Charger pour charger d’abord un fichier de schéma. Pour en savoir plus, voir Comprendre les schémas de messages.

Bicep

Une fois que vous avez utilisé le registre de schémas pour stocker le schéma, vous pouvez le référencer dans la configuration du flux de données.

sourceSettings: {
  serializationFormat: 'Json'
  schemaRef: 'aio-sr://<SCHEMA_NAMESPACE>/<SCHEMA_NAME>:<VERSION>'
}

Kubernetes

Une fois que vous avez utilisé le registre de schémas pour stocker le schéma, vous pouvez le référencer dans la configuration du flux de données.

sourceSettings:
  serializationFormat: Json
  schemaRef: 'aio-sr://<SCHEMA_NAMESPACE>/<SCHEMA_NAME>:<VERSION>'

Transformation

L’opération de transformation a lieu là où vous pouvez transformer les données à la source avant de les envoyer à la destination. Les transformations sont facultatives. Si vous n’avez pas besoin d’apporter de modifications aux données, n’incluez pas l’opération de transformation dans la configuration du flux de données. Plusieurs transformations sont reliées sous forme de phases, quel que soit leur ordre dans la configuration. L’ordre des étapes est toujours :

1. Enrichissez, renommez ou ajoutez une Nouvelle propriété : ajoutez des données supplémentaires aux données sources en fonction d’un jeu de données et d’une condition à mettre en correspondance.
2. Filtrer : filtrer les données en fonction d’une condition.
3. Mapper ou Calculer : déplacer des données d’un champ vers un autre avec une conversion facultative.

Portail

Dans le portail d’expérience des opérations, sélectionnez Flux de données>Ajouter une transformation (facultatif).

Bicep

builtInTransformationSettings: {
  datasets: [
    // See section on enriching data
  ]
  filter: [
    // See section on filtering data
  ]
  map: [
    // See section on mapping data
  ]
}

Kubernetes

builtInTransformationSettings:
  datasets:
    # See section on enriching data
  filter:
    # See section on filtering data
  map:
    # See section on mapping data

Enrichir : Ajouter des données de référence

Pour enrichir les données, vous pouvez utiliser un jeu de données de référence dans le magasin d’états distribué (DSS) d’Opérations Azure IoT. Le jeu de données est utilisé pour ajouter des données supplémentaires aux données sources en fonction d’une condition. La condition est spécifiée en tant que champ dans les données sources qui correspond à un champ du jeu de données.

Vous pouvez charger des exemples de données dans la DSS à l’aide de l’exemple d’outil de jeu DSS. Les noms de clés dans le magasin d’états distribué correspondent à un jeu de données dans la configuration du flux de données.

Portail

Dans l’expérience des opérations, l’étape d’enrichissement est actuellement prise en charge à l’aide des transformations Renommer and Nouvelle propriété.

1. Dans le portail d’expérience des opérations, sélectionnez Ajouter une transformation (facultatif).

2. Choisissez Renommer ou Nouvelle propriété, puis sélectionnez Ajouter.

   

Bicep

Cet exemple montre comment vous pouvez utiliser le champ deviceId dans les données sources pour faire correspondre le champ asset dans le jeu de données :

builtInTransformationSettings: {
  datasets: [
    {
      key: 'assetDataset'
      inputs: [
        '$source.deviceId' // ---------------- $1
        '$context(assetDataset).asset' // ---- $2
      ]
      expression: '$1 == $2'
    }
  ]
}

Kubernetes

Par exemple, vous pouvez utiliser le champ deviceId dans les données sources pour faire correspondre le champ asset dans le jeu de données :

builtInTransformationSettings:
  datasets:
  - key: assetDataset
    inputs:
      - $source.deviceId # ------------- $1
      - $context(assetDataset).asset # - $2
    expression: $1 == $2

Si le jeu de données a un enregistrement avec le champ asset similaire à :

{
  "asset": "thermostat1",
  "location": "room1",
  "manufacturer": "Contoso"
}

Les données de la source avec le champ deviceId correspondant à thermostat1 ont les champs location et manufacturer disponibles dans les phases de filtre et de carte.

Pour plus d’informations sur la syntaxe des conditions, consultez Enrichir des données à l’aide de flux de données et Convertir des données à l’aide de flux de données.

Filtrer : Filtrer les données en fonction d’une condition

Pour filtrer les données selon une condition, vous pouvez utiliser la phase filter. La condition est spécifiée en tant que champ dans les données sources qui correspond à une valeur.

Portail

1. Sous Transformation (facultatif), sélectionnez Filtrer>Ajouter.

2. Choisissez les points de données à inclure dans le jeu de données.

3. Ajoutez une condition de filtre et une description.

   

4. Sélectionnez Appliquer.

Par exemple, vous pouvez utiliser une condition de filtre comme temperature > 20 pour filtrer les données inférieures ou égales à 20 en fonction du champ de température.

Bicep

Par exemple, vous pouvez utiliser le champ temperature dans les données sources pour filtrer les données :

builtInTransformationSettings: {
  filter: [
    {
      inputs: [
        'temperature ? $last'
      ]
      expression: '$1 > 20'
    }
  ]
}

Si le champ temperature est supérieur à 20, les données sont passées à la phase suivante. Si le champ temperature est inférieur ou égal à 20, les données sont filtrées.

Kubernetes

Par exemple, vous pouvez utiliser le champ temperature dans les données sources pour filtrer les données :

builtInTransformationSettings:
  filter:
    - inputs:
      - temperature ? $last # - $1
      expression: "$1 > 20"

Si le champ temperature est supérieur à 20, les données sont passées à la phase suivante. Si le champ temperature est inférieur ou égal à 20, les données sont filtrées.

Mapper : Déplacer les données d’un champ vers un autre

Pour mapper les données à un autre champ avec une conversion facultative, vous pouvez utiliser l’opération map. La conversion est spécifiée en tant que formule qui utilise les champs dans les données sources.

Portail

Dans l’expérience des opérations, le mappage est actuellement pris en charge à l’aide de transformations de calcul.

1. Sous Transformation (facultatif), sélectionnez Calcul>Ajouter.

2. Entrez les champs et expressions obligatoires.

   

3. Sélectionnez Appliquer.

Bicep

Par exemple, vous pouvez utiliser le champ temperature dans les données sources pour convertir la température en degrés Celsius et la stocker dans le champ temperatureCelsius. Vous pouvez également enrichir les données sources avec le champ location à partir du jeu de données de contextualisation :

builtInTransformationSettings: {
  map: [
    {
      inputs: [
        'temperature'
      ]
      output: 'temperatureCelsius'
      expression: '($1 - 32) * 5/9'
    }
    {
      inputs: [
        '$context(assetDataset).location'
      ]
      output: 'location'
    }
  ]
}

Kubernetes

Par exemple, vous pouvez utiliser le champ temperature dans les données sources pour convertir la température en degrés Celsius et la stocker dans le champ temperatureCelsius. Vous pouvez également enrichir les données sources avec le champ location à partir du jeu de données de contextualisation :

builtInTransformationSettings:
  map:
    - inputs:
      - temperature # - $1
      expression: "($1 - 32) * 5/9"
      output: temperatureCelsius
    - inputs:
      - $context(assetDataset).location  
      output: location

Pour en savoir plus, consultez Mapper des données à l’aide de flux de données et Convertir des données à l’aide de flux de données.

Sérialiser des données en fonction d’un schéma

Si vous souhaitez sérialiser les données avant de les envoyer à la destination, vous devez spécifier un schéma et un format de sérialisation. Sinon, les données sont sérialisées au format JSON avec les types inférés. Les points de terminaison de stockage tels que Microsoft Fabric ou Azure Data Lake nécessitent un schéma pour garantir la cohérence des données. Les formats de sérialisation pris en charge sont Parquet et Delta.

Portail

Actuellement, la spécification du schéma de sortie et de la sérialisation n’est pas prise en charge dans l’expérience des opérations.

Bicep

Une fois que vous avez chargé un schéma dans le registre des schémas, vous pouvez le référencer dans la configuration du flux de données.

builtInTransformationSettings: {
  serializationFormat: 'Delta'
  schemaRef: 'aio-sr://<SCHEMA_NAMESPACE>/<SCHEMA>:<VERSION>'
}

Kubernetes

Une fois que vous avez chargé un schéma dans le registre des schémas, vous pouvez le référencer dans la configuration du flux de données.

builtInTransformationSettings:
  serializationFormat: Delta
  schemaRef: 'aio-sr://<SCHEMA_NAMESPACE>/<SCHEMA>:<VERSION>'

Pour plus d’informations sur le registre de schémas, consultez Comprendre les schémas de message.

Destination

Pour configurer une destination pour le flux de données, spécifiez la référence de point de terminaison et la destination de données. Vous pouvez spécifier une liste de destinations de données pour le point de terminaison.

Pour envoyer des données à une destination autre que le répartiteur MQTT local, créez un point de terminaison de flux de données. Pour en savoir plus, voir Configurer les points de terminaison du flux de données. Si la destination n’est pas l’Agent MQTT local, il doit être utilisé comme source. Pour en savoir plus, consultez Dataflows doit utiliser le point de terminaison de l’Agent MQTT local.

Important

Les points de terminaison de stockage nécessitent une référence de schéma. Si vous avez créé des points de terminaison de destination de stockage pour Microsoft Fabric OneLake, ADLS Gen 2, Azure Data Explorer et Stockage local, vous devez spécifier la référence de schéma.

Portail

1. Sélectionnez le point de terminaison de flux de données à utiliser comme destination.

   

2. Sélectionnez Continuer pour configurer la destination.

3. Entrez les paramètres requis pour la destination, y compris la rubrique ou la table à laquelle envoyer les données. Pour plus d’informations, consultez Configurer la destination des données (rubrique, conteneur ou table ).

Bicep

destinationSettings: {
  endpointRef: '<CUSTOM_ENDPOINT_NAME>'
  dataDestination: '<TOPIC_OR_TABLE>' // See section on configuring data destination
}

Kubernetes

destinationSettings:
  endpointRef: <CUSTOM_ENDPOINT_NAME>
  dataDestination: <TOPIC_OR_TABLE> # See section on configuring data destination

Configurer la destination des données (rubrique, conteneur ou table)

Comme pour les sources de données, la destination de données est un concept utilisé pour conserver les points de terminaison de flux de données réutilisables sur plusieurs flux de données. Essentiellement, il représente le sous-répertoire dans la configuration du point de terminaison de flux de données. Par exemple, si le point de terminaison de flux de données est un point de terminaison de stockage, la destination des données est la table du compte de stockage. Si le point de terminaison de flux de données est un point de terminaison Kafka, la destination de données est la rubrique Kafka.

Type de point de terminaison	Sens de destination des données	Description
MQTT (ou Event Grid)	Sujet	Rubrique MQTT dans laquelle les données sont envoyées. Seules les rubriques statiques sont prises en charge, pas de caractères génériques.
Kafka (ou Event Hubs)	Sujet	Rubrique Kafka dans laquelle les données sont envoyées. Seules les rubriques statiques sont prises en charge, pas de caractères génériques. Si le point de terminaison est un espace de noms Event Hubs, la destination des données est le hub d’événements individuel dans l’espace de noms.
Azure Data Lake Storage	Conteneur	Le conteneur dans le compte de stockage. Pas la table.
Microsoft Fabric OneLake	Table ou dossier	Correspond au type de chemin d’accès configuré pour le point de terminaison.
Explorateur de données Azure	Table	La table dans la base de données Azure Data Explorer.
Stockage local	Dossier	Nom du dossier ou du répertoire dans le montage de volume persistant de stockage local. Lors de l’utilisation d’Azure Container Storage activée par Azure Arc Cloud Ingest Edge Volumes, cela doit correspondre au paramètre spec.path de la sous-valeur que vous avez créée.

Pour configurer la destination des données :

Portail

Lorsque vous utilisez l’expérience des opérations, le champ de destination de données est automatiquement interprété en fonction du type de point de terminaison. Par exemple, si le point de terminaison de flux de données est un point de terminaison de stockage, la page de détails de destination vous invite à entrer le nom du conteneur. Si le point de terminaison de flux de données est un point de terminaison MQTT, la page des détails de destination vous invite à entrer la rubrique, et ainsi de suite.

Bicep

La syntaxe est la même pour tous les points de terminaison de flux de données :

destinationSettings: {
  endpointRef: "<CUSTOM_ENDPOINT_NAME>"
  dataDestination: '<TOPIC_OR_TABLE>'
}

Par exemple, pour renvoyer des données au répartiteur MQTT local, utilisez la configuration suivante :

destinationSettings: {
  endpointRef: 'default'
  dataDestination: 'example-topic'
}

Ou, si vous disposez d’un point de terminaison Event Hub personnalisé, la configuration se présente comme suit :

destinationSettings: {
  endpointRef: 'my-eh-endpoint'
  dataDestination: 'individual-event-hub'
}

Un autre exemple utilisant un point de terminaison de stockage comme destination :

destinationSettings: {
  endpointRef: 'my-adls-endpoint'
  dataDestination: 'my-container'
}

Kubernetes

La syntaxe est la même pour tous les points de terminaison de flux de données :

destinationSettings:
  endpointRef: <CUSTOM_ENDPOINT_NAME>
  dataDestination: <TOPIC_OR_TABLE>

Par exemple, pour renvoyer des données au répartiteur MQTT local, utilisez la configuration suivante :

destinationSettings:
  endpointRef: default
  dataDestination: example-topic

Ou, si vous disposez d’un point de terminaison Event Hub personnalisé, la configuration se présente comme suit :

destinationSettings:
  endpointRef: my-eh-endpoint
  dataDestination: individual-event-hub

Un autre exemple utilisant un point de terminaison de stockage comme destination :

destinationSettings:
  endpointRef: my-adls-endpoint
  dataDestination: my-container

Exemple

L’exemple suivant illustre une configuration de flux de données utilisant le point de terminaison MQTT comme source et destination. La source filtre les données de la rubriques MQTT azure-iot-operations/data/thermostat. La transformation convertit la température en Fahrenheit et filtre les données où la température multipliée par l’humidité est inférieure à 100 000. La destination envoie les données à la rubrique MQTT factory.

Portail

Consultez les onglets Bicep ou Kubernetes pour obtenir l’exemple de configuration.

Bicep

param aioInstanceName string = '<AIO_INSTANCE_NAME>'
param customLocationName string = '<CUSTOM_LOCATION_NAME>'
param dataflowName string = '<DATAFLOW_NAME>'

resource aioInstance 'Microsoft.IoTOperations/instances@2024-09-15-preview' existing = {
  name: aioInstanceName
}

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

// Pointer to the default dataflow endpoint
resource defaultDataflowEndpoint 'Microsoft.IoTOperations/instances/dataflowEndpoints@2024-09-15-preview' existing = {
  parent: aioInstance
  name: 'default'
}

// Pointer to the default dataflow profile
resource defaultDataflowProfile 'Microsoft.IoTOperations/instances/dataflowProfiles@2024-09-15-preview' existing = {
  parent: aioInstance
  name: 'default'
}

resource dataflow 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflows@2024-09-15-preview' = {
  // Reference to the parent dataflow profile, the default profile in this case
  // Same usage as profileRef in Kubernetes YAML
  parent: defaultDataflowProfile
  name: dataflowName
  extendedLocation: {
    name: customLocation.id
    type: 'CustomLocation'
  }
  properties: {
    mode: 'Enabled'
    operations: [
      {
        operationType: 'Source'
        sourceSettings: {
          // Use the default MQTT endpoint as the source
          endpointRef: defaultDataflowEndpoint.name
          // Filter the data from the MQTT topic azure-iot-operations/data/thermostat
          dataSources: [
            'azure-iot-operations/data/thermostat'
          ]
        }
      }
      // Transformation optional
      {
        operationType: 'BuiltInTransformation'
        builtInTransformationSettings: {
          // Filter the data where temperature * "Tag 10" < 100000
          filter: [
            {
              inputs: [
                'temperature.Value'
                '"Tag 10".Value'
              ]
              expression: '$1 * $2 < 100000'
            }
          ]
          map: [
            // Passthrough all values by default
            {
              inputs: [
                '*'
              ]
              output: '*'
            }
            // Convert temperature to Fahrenheit and output it to TemperatureF
            {
              inputs: [
                'temperature.Value'
              ]
              output: 'TemperatureF'
              expression: 'cToF($1)'
            }
          // Extract the "Tag 10" value and output it to Humidity
            {
              inputs: [
                '"Tag 10".Value'
              ]
              output: 'Humidity'
            }
          ]
        }
      }
      {
        operationType: 'Destination'
        destinationSettings: {
          // Use the default MQTT endpoint as the destination
          endpointRef: defaultDataflowEndpoint.name
          // Send the data to the MQTT topic factory
          dataDestination: 'factory'
        }
      }
    ]
  }
}

Kubernetes

apiVersion: connectivity.iotoperations.azure.com/v1beta1
kind: Dataflow
metadata:
  name: my-dataflow
  namespace: azure-iot-operations
spec:
  # Reference to the default dataflow profile
  profileRef: default
  mode: Enabled
  operations:
    - operationType: Source
      sourceSettings:
        # Use the default MQTT endpoint as the source
        endpointRef: default
        # Filter the data from the MQTT topic azure-iot-operations/data/thermostat
        dataSources:
          - azure-iot-operations/data/thermostat
    # Transformation optional
    - operationType: builtInTransformation
      builtInTransformationSettings:
        # Filter the data where temperature * "Tag 10" < 100000
        filter:
          - inputs:
              - 'temperature.Value'
              - '"Tag 10".Value'
            expression: '$1 * $2 < 100000'
        map:
          # Passthrough all values by default
          - inputs:
              - '*'
            output: '*'
          # Convert temperature to Fahrenheit and output it to TemperatureF
          - inputs:
              - temperature.Value
            output: TemperatureF
            expression: cToF($1)
          # Extract the "Tag 10" value and output it to Humidity
          - inputs:
              - '"Tag 10".Value'
            output: 'Humidity'
    - operationType: Destination
      destinationSettings:
        # Use the default MQTT endpoint as the destination
        endpointRef: default
        # Send the data to the MQTT topic factory
        dataDestination: factory

Pour obtenir d’autres exemples de configurations de flux de données, consultez API REST Azure : flux de données et le démarrage rapide Bicep.

Vérifier qu’un flux de données fonctionne

Suivez le Tutoriel : pont MQTT bidirectionnel vers Azure Event Grid pour vérifier que le flux de données fonctionne.

Exporter la configuration du flux de données

Pour exporter la configuration du flux de données, vous pouvez utiliser l’expérience des opérations ou en exportant la ressource personnalisée Flux de données.

Portail

Sélectionnez le flux de données que vous souhaitez exporter et sélectionnez Exporter dans la barre d’outils.

Bicep

Bicep est une infrastructure en tant que code et aucune exportation n’est requise. Utilisez le fichier de modèle Bicep pour créer un flux de données pour configurer et configurer rapidement des flux de données.

Kubernetes

kubectl get dataflow my-dataflow -o yaml > my-dataflow.yaml

Configuration appropriée du flux de données

Pour vous assurer que le flux de données fonctionne comme prévu, vérifiez les éléments suivants :

• Le point de terminaison du flux de données MQTT par défaut doit être utilisé comme source ou comme destination.
• Le profil de flux de données existe et est référencé dans la configuration du flux de données.
• La source est un point de terminaison MQTT, un point de terminaison Kafka ou une ressource. Les points de terminaison de type stockage ne peuvent pas être utilisés comme source.
• Lorsque vous utilisez Event Grid comme source, le nombre d’instances de profil de flux de données est défini sur 1, car l’Agent MQTT Event Grid ne prend pas en charge les abonnements partagés.
• Lorsque vous utilisez Event Hubs comme source, chaque Event Hub de l’espace de noms est une rubrique Kafka distincte et doit être spécifié comme la source de données.
• La transformation, si elle est utilisée, est configurée avec une syntaxe appropriée, y compris l’échappement de caractères spéciaux approprié.
• Lorsque vous utilisez des points de terminaison de type stockage comme destination, un schéma est spécifié.

Étapes suivantes

• Mapper des données à l’aide de flux de données
• Convertir des données à l’aide de flux de données
• Enrichir des données à l’aide de flux de données
• Comprendre les schémas de message
• Gérer le profil de flux de données