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.

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

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

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

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

resource dataflow 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflows@2024-08-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
        }
      }
      {
        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
    - 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 en tant que 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.

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: defaultDataflowEndpoint
  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. Cela 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.

Pour plus d’informations sur le point de terminaison MQTT par défaut et la création d’un point de terminaison MQTT en tant que source de flux de données, consultez point de terminaison MQTT.

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. Cela 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.

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

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>

Remarque

Si le nombre d’instances dans le profil de flux de données est supérieur à 1, la rubrique d’abonnement partagé est automatiquement ajouté au filtre de rubrique.

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. 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).

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

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

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'
    }
  ]
}

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.

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.

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.

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 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: default
        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: default
        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 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

É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