Tutoriel : acheminer des données en utilisant des stratégies d'actualisation de tables

Changer de service à l’aide de la liste déroulante Version . En savoir plus sur la navigation.
S’applique à : ✅ Microsoft Fabric ✅ Azure Data Explorer

Lorsque vos données sources impliquent des transformations simples et rapides, effectuez-les en amont dans le pipeline à l’aide d’un flux d’événements. Toutefois, cette approche peut ne pas fonctionner correctement pour d’autres transformations complexes ou nécessitant une fonctionnalité spécialisée pour fonctionner.

Dans ce tutoriel, vous allez apprendre à :

L’exemple de ce didacticiel montre comment utiliser des stratégies de mise à jour pour le routage des données afin d’effectuer des transformations complexes pour enrichir, nettoyer et transformer des données au moment de l’ingestion. Pour obtenir la liste des autres cas d’usage courants, consultez Les cas d’usage courants pour les stratégies de mise à jour de table.

Prerequisites

  • Un compte Microsoft ou une identité utilisateur Microsoft Entra. Un abonnement Azure n’est pas requis.

Étape 1 : Créer des tables et mettre à jour des stratégies

Les étapes suivantes vous guident tout au long de la création d’une table source, de fonctions de transformation, de tables de destination et de stratégies de mise à jour. Le tutoriel montre comment utiliser des stratégies de mise à jour de table pour effectuer des transformations complexes et enregistrer les résultats dans une ou plusieurs tables de destination. L’exemple utilise une table source unique nommée Raw_Table et trois tables de destination nommées Device_Telemetry, Device_Alarms et Error_Log.

  1. Exécutez la commande suivante pour créer une table nommée Raw_Table.

    .create table Raw_Table (RawData: dynamic)

    La table source est l’emplacement où les données ingérées sont enregistrées. La table comporte une seule colonne nommée RawData de type dynamique. Le type dynamique stocke les données brutes telles quelles, sans schéma. Pour plus d’informations, consultez la commande .create table.

  2. Exécutez la commande suivante pour créer des fonctions nommées Get_Telemetry, Get_Alarms et Log_Error.

    .execute database script <|
  .create-or-alter function Get_Telemetry() {
    Raw_Table
    | where todynamic(RawData).MessageType == 'Telemetry'
    | extend
      Timestamp = unixtime_seconds_todatetime(tolong(RawData.Timestamp)),
      DeviceId = tostring(RawData.DeviceId),
      DeviceType = tostring(RawData.DeviceType),
      SensorName = tostring(RawData.SensorName),
      SensorValue = toreal(RawData.SensorValue),
      SensorUnit = tostring(RawData.SensorUnit)
    | project-away RawData
  }
  .create-or-alter function Get_Alarms() {
    Raw_Table
    | where RawData.MessageType == 'Alarms'
    | extend
      Timestamp = unixtime_seconds_todatetime(tolong(RawData.Timestamp)),
      DeviceId = tostring(RawData.DeviceId),
      DeviceType = tostring(RawData.DeviceTpe) ,
      AlarmType = tostring(RawData.AlarmType)
    | project-away RawData
  }
  .create-or-alter function Log_Error() {
    Raw_Table
    | where RawData.MessageType !in ('Telemetry', 'Alarms')
    | extend
      TimeStamp = datetime(now),
      ErrorType = 'Unknown MessageType'
    | project TimeStamp, RawData, ErrorType
  }

    Lors de la création d’une stratégie de mise à jour, vous pouvez spécifier un script inline pour l’exécution. Toutefois, encapsulez la logique de transformation dans une fonction. L’utilisation d’une fonction améliore la maintenance du code. Lorsque de nouvelles données arrivent, la fonction s’exécute pour transformer les données. La fonction peut être réutilisée dans plusieurs stratégies de mise à jour. Pour plus d’informations, consultez la commande .create function.

  3. Exécutez la commande suivante pour créer les tables de destination.

    .execute database script <|
  .create table Device_Telemetry (Timestamp: datetime, DeviceId: string, DeviceType: string, SensorName: string, SensorValue: real, SensorUnit: string)
  .set-or-append Device_Alarms <| Get_Alarms | take 0
  .set-or-append Error_Log <| Log_Error | take 0

    La table de destination doit avoir le même schéma que la sortie de la fonction de transformation. Vous pouvez créer des tables de destination de la manière suivante :

    • Utilisez la .create table commande et spécifiez manuellement le schéma comme illustré avec la création de la table Device_Telemetry . Toutefois, cette approche peut être sujette aux erreurs et fastidieuse.
    • Utilisez la .set-or-append commande si vous avez déjà créé une fonction pour transformer les données. Cette méthode crée une table avec le même schéma que la sortie de la fonction, en utilisant take 0 pour vous assurer que la fonction retourne uniquement le schéma. Pour plus d’informations, consultez la commande .set-or-append.

  4. Exécutez la commande suivante pour créer les stratégies de mise à jour pour les tables de destination.

    .execute database script <|
  .alter table Device_Telemetry policy update "[{\"IsEnabled\":true,\"Source\":\"Raw_Table\",\"Query\":\"Get_Telemetry\",\"IsTransactional\":false,\"PropagateIngestionProperties\":true,\"ManagedIdentity\":null}]"
  .alter table Device_Alarms policy update "[{\"IsEnabled\":true,\"Source\":\"Raw_Table\",\"Query\":\"Get_Alarms\",\"IsTransactional\":false,\"PropagateIngestionProperties\":true,\"ManagedIdentity\":null}]"
  .alter table Error_Log policy update "[{\"IsEnabled\":true,\"Source\":\"Raw_Table\",\"Query\":\"Log_Error\",\"IsTransactional\":false,\"PropagateIngestionProperties\":true,\"ManagedIdentity\":null}]"

    Utilisez la commande pour lier la .alter table policy update table source, la fonction de transformation et la table de destination. Créez la stratégie de mise à jour sur la table de destination et spécifiez la table source et la fonction de transformation. Pour plus d’informations, consultez la commande .alter table policy update.

Étape 2 : Ingérer des exemples de données

Pour tester les stratégies de mise à jour, ingérer des exemples de données dans la table source à l’aide de la .set-or-append commande. Pour plus d’informations, consultez Ingestion de données à partir d’une requête.

.set-or-append Raw_Table <|
  let Raw_Stream = datatable(RawData: dynamic)
    [
    dynamic({"TimeStamp": 1691757932, "DeviceId": "Sensor01", "MessageType": "Telemetry", "DeviceType": "Laminator", "SensorName": "Temperature", "SensorValue": 78.3, "SensorUnit": "Celcius"}),
    dynamic({"TimeStamp": 1691757932, "DeviceId": "Sensor01", "MessageType": "Alarms", "DeviceType": "Laminator", "AlarmType": "Temperature threshold breached"}),
    dynamic({"TimeStamp": 1691757932, "DeviceId": "Sensor01", "MessageType": "Foo", "ErrorType": "Unknown"})
  ];
  Raw_Stream

Étape 3 : vérifier les résultats

Pour valider les résultats, exécutez une requête pour vérifier que les données ont été transformées et routées vers les tables de destination. Dans l’exemple suivant, l’opérateur union combine la source et les résultats des tables de destination dans un jeu de résultats unique.

Raw_Table | summarize Rows=count() by TableName = "Raw_Table"
| union (Device_Telemetry | summarize Rows=count() by TableName = "Device_Telemetry")
| union (Device_Alarms | summarize Rows=count() by TableName = "Device_Alarms")
| union (Error_Log | summarize Rows=count() by TableName = "Error_Log")
| sort by Rows desc

Output

Vous devez voir la sortie suivante où le Raw_Table a trois lignes et les tables de destination ont chacune une ligne.

TableName Rows
Raw_Table 3
Journal_d'erreurs 1
Alarmes_de_l'appareil 1
Télémétrie de dispositif 1

Étape 4 : Nettoyer les ressources

Exécutez la commande suivante dans votre base de données pour nettoyer les tables et les fonctions créées dans ce didacticiel.

.execute database script <|
  .drop table Raw_Table
  .drop table Device_Telemetry
  .drop table Device_Alarms
  .drop table Error_Log
  .drop function Get_Telemetry
  .drop function Get_Alarms
  .drop function Log_Error

Contenu connexe

  • Last updated on