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.

Cet article vous montre comment créer un flux de données avec un exemple, notamment la source, la transformation et la destination.

Prérequis

• Une instance d’Opérations Azure IoT (préversion)
• Un profil de flux de données configuré
• Points de terminaison de flux de données. Par exemple, créez un point de terminaison de flux de données pour l’Agent MQTT local. Vous pouvez utiliser ce point de terminaison pour la source et la destination. Vous pouvez également essayer d’autres points de terminaison 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.

Créer un flux de données

Une fois que vous avez des points de terminaison de flux de données, vous pouvez les utiliser pour créer un flux de données. Rappelez-vous qu’un flux de données est constitué de trois parties : la source, la transformation et la destination.

Portail

Pour créer un flux de données dans le portail d’expérience des opérations, sélectionnez Flux de données>Créer un flux de données.

Kubernetes

La structure globale d’une configuration de flux de données est la suivante :

apiVersion: connectivity.iotoperations.azure.com/v1beta1
kind: Dataflow
metadata:
  name: my-dataflow
  namespace: azure-iot-operations
spec:
  profileRef: default
  mode: Enabled
  operations:
    - operationType: Source
      sourceSettings:
        # See source configuration section
    - 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.

Configurer une source avec un point de terminaison de flux de données pour obtenir des données

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

Portail

Utiliser la ressource en tant que source

Vous pouvez utiliser une ressource comme source du flux de données. Cela est disponible uniquement dans le portail d’expérience des opérations.

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.

Kubernetes

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

Utiliser MQTT comme source

Portail

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

2. Entrez la rubrique MQTT à écouter pour les messages entrants.

3. Choisissez un schéma de message dans la liste déroulante ou chargez un nouveau schéma. 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.

   

4. Sélectionnez Appliquer.

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: mq
  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 plus d’informations, consultez Réutiliser les points de terminaison de flux de données.

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.

spec:
  operations:
  - operationType: Source
    sourceSettings:
      serializationFormat: Json
      schemaRef: aio-sr://exampleNamespace/exampleAvroSchema:1.0.0

Pour spécifier le schéma, créez le fichier et stockez-le dans le registre de schémas.

{
  "type": "record",
  "name": "Temperature",
  "fields": [
    {"name": "deviceId", "type": "string"},
    {"name": "temperature", "type": "float"}
  ]
}

Remarque

Le seul format de sérialisation pris en charge est JSON. Le schéma est facultatif.

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

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/<subscription-group>/<topic>.

sourceSettings:
  dataSources:
    - $shared/myGroup/thermostats/+/telemetry/temperature/#

Remarque

Si le nombre d’instances dans le profil de flux de données est supérieur à 1, la rubrique d’abonnement partagé doit être utilisée.

Configurer la transformation pour traiter les données

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. Enrichir : ajouter des données supplémentaires aux données sources en fonction d’un jeu de données et d’une condition à remplir.
2. Filtrer : filtrer les données en fonction d’une condition.
3. Mapper : 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).

Kubernetes

builtInTransformationSettings:
  datasets:
    # ...
  filter:
    # ...
  map:
    # ...

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.

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

Actuellement, l’opération d’enrichissement n’est pas disponible dans le portail d’expérience des opérations.

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 filter et map.

Vous pouvez charger des exemples de données dans la DSS à l’aide de l’exemple d’outil de jeu DSS.

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.

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 le portail d’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.

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
      output: temperatureCelsius
      expression: "($1 - 32) * 5/9"
    - 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. N’oubliez pas que 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.

Portail

Spécifiez le schéma sortie lorsque vous ajoutez le point de terminaison de flux de données de destination.

Kubernetes

builtInTransformationSettings:
  serializationFormat: Parquet
  schemaRef: aio-sr://<NAMESPACE>/<SCHEMA>:<VERSION>

Pour spécifier le schéma, vous pouvez créer une ressource personnalisée de schéma avec la définition de schéma.

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

{
  "type": "record",
  "name": "Temperature",
  "fields": [
    {"name": "deviceId", "type": "string"},
    {"name": "temperatureCelsius", "type": "float"}
    {"name": "location", "type": "string"}
  ]
}

Les formats de sérialisation pris en charge sont JSON, Parquet et Delta.

Configurer la destination avec un point de terminaison de flux de données pour envoyer des données

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 qui sont des rubriques MQTT ou Kafka.

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. Ajoutez les détails de mappage en fonction du type de destination.

Kubernetes

Par exemple, pour configurer une destination à l’aide du point de terminaison MQTT créé précédemment et d’une rubrique MQTT statique, utilisez la configuration suivante :

destinationSettings:
  endpointRef: mq
  dataDestination: factory

Si vous avez créé des points de terminaison de stockage comme Microsoft Fabric, utilisez le champ de destination des données pour spécifier le nom de la table ou du conteneur :

destinationSettings:
  endpointRef: adls
  dataDestination: telemetryTable

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 des rubriques MQTT thermostats/+/telemetry/temperature/# et humidifiers/+/telemetry/humidity/#. La transformation convertit la température en Fahrenheit et filtre les données où la température est inférieure à 10 000. La destination envoie les données à la rubrique MQTT factory.

apiVersion: connectivity.iotoperations.azure.com/v1beta1
kind: Dataflow
metadata:
  name: my-dataflow
  namespace: azure-iot-operations
spec:
  profileRef: default
  mode: Enabled
  operations:
    - operationType: Source
      sourceSettings:
        endpointRef: mq
        dataSources:
          - thermostats/+/telemetry/temperature/#
          - humidifiers/+/telemetry/humidity/#
    - operationType: builtInTransformation
      builtInTransformationSettings:
        filter:
          - inputs:
              - 'temperature.Value'
              - '"Tag 10".Value'
            expression: "$1*$2<100000"
        map:
          - inputs:
              - '*'
            output: '*'
          - inputs:
              - temperature.Value
            output: TemperatureF
            expression: cToF($1)
          - inputs:
              - '"Tag 10".Value'
            output: 'Tag 10'
    - operationType: Destination
      destinationSettings:
        endpointRef: mq
        dataDestination: factory

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 le portail d’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.

Kubernetes

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