Partager via

Utiliser les dernières valeurs connues dans un pipeline Processeur de données Azure IoT préversion

  • Article

Important

Opérations Azure IoT (préversion) – activé parc Azure Arc est actuellement en PRÉVERSION. Vous ne devez pas utiliser ce logiciel en préversion dans des environnements de production.

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.

Utilisez l’index de dernières valeurs connues (LKV) dans un pipeline de processeur de données pour maintenir un enregistrement à jour et complet de vos données. L’index LKV suit les dernières valeurs des paires clés-valeurs pour les messages dans le pipeline. L’index peut ensuite enrichir les messages à l’aide des valeurs LKV suivies. Le suivi et l’enrichissement des dernières valeurs connues sont importants pour les processus en aval qui reposent sur :

  • plusieurs points de données de série chronologique à un timestamp spécifique ;
  • des charges utiles qui ont toujours une valeur pour une clé particulière.

Dans un pipeline de traitement des données, l’index LKV est facultatif. Lorsque vous utilisez l’index LKV, vous pouvez :

  • ajouter plusieurs index LKV à un pipeline. Chaque index LKV peut suivre plusieurs valeurs ;
  • enrichir les messages avec les valeurs LKV stockées, ce qui garantit que les données demeurent complètes ;
  • conserver automatiquement les valeurs LKV mises à jour avec les valeurs les plus récentes des messages entrants ;
  • suivre les valeurs LKV séparément pour chaque partition logique L’index LKV fonctionne de manière indépendante dans chaque partition logique.
  • Configurez l’heure d’expiration de chaque valeur LKV suivie afin de gérer la durée pendant laquelle elle reste valide. Ce contrôle permet de garantir que les messages ne sont pas enrichis avec des valeurs obsolètes.

L’index LKV maintient l’intégrité chronologique des données. L’index garantit que les messages avec des timestamps antérieurs ne remplacent pas les valeurs LKV par des messages qui ont des timestamps ultérieurs.

L’index LKV enrichit les messages entrants avec les dernières valeurs connues qu’il suit. Ces valeurs enrichies représentent des données enregistrées précédemment et ne constituent pas nécessairement les valeurs en temps réel actuelles. Assurez-vous que ce comportement s’aligne sur vos attentes en matière de traitement des données.

Prérequis

Pour configurer et utiliser une phase de pipeline agrégé, vous aurez besoin d’une instance déployée du processeur de données Azure IoT Data Processor – Préversion qui inclut le composant Data Processor en option.

Configurer l’index

La configuration JSON de l’index LKV définit les détails de l’index. Pour créer l’index, vous pouvez interagir avec l’interface utilisateur basée sur le formulaire ou fournir la configuration JSON sous l’onglet Advanced (Avancé) :

Champ Description Obligatoire Par défaut Exemple
Nom Nom de l’index défini par l’utilisateur. Oui - lkv1
Description Description de l’index définie par l’utilisateur. Non - lkv1
Chemin d’accès entrant > des propriétés Le chemin d’accès de la clé à suivre. Oui - .payload.temperature
Chemin d’accès sortant > des propriétés Le chemin d’accès à l’emplacement dans le message sortant pour écrire la valeur LKV. Oui - .payload.temperature_lkv
Durée d’expiration > des propriétés Les LKV suivies sont valides uniquement pour l’intervalle de temps défini par l’utilisateur. Au terme de ce délai, le message sortant n’est pas enrichi avec la valeur stockée. L’expiration est suivie pour chaque clé LKV. Non - 10h
Chemin d’accès de timestamp > des propriétés Le chemin d’accès à l’emplacement dans le message sortant pour écrire le timestamp de la dernière mise à jour de la valeur LKV. Non False -

L’inclusion du chemin d’accès du timestamp vous aide à comprendre précisément quand les valeurs LKV ont été enregistrées, et améliore la transparence et la traçabilité.

inputPath est égal à outputPath

Le message de sortie est soit la valeur réelle du message, soit la valeur LKV si la clé suivie est manquante dans la charge utile du message. Toute valeur entrante est prioritaire et l’index ne la remplace pas par une valeur LKV. Pour identifier si la valeur du message est une valeur LKV, utilisez le chemin d’accès du timestamp. Le chemin d’accès du timestamp est inclus uniquement dans le message sortant si la valeur du message est la valeur LKV suivie.

inputPath n’est pas égal à outputPath

L’index écrit la valeur LKV dans outputPath pour tous les messages entrants. Utilisez cette configuration pour suivre la différence entre les valeurs dans les charges utiles de messages suivantes.

Exemple de configuration

L’exemple suivant illustre un message pour l’index LKV avec le message arrivant à 10:02 et une charge utile qui contient la valeur LKV suivie .payload.temperature :

{ 
  { 
    "systemProperties":{ 
        "partitionKey":"pump", 
        "partitionId":5, 
        "timestamp":"2023-01-11T10:02:07Z" 
    }, 
    "qos":1, 
    "topic":"/assets/pump/#" 
    }, 
    "payload":{ 
        "humidity": 10, 
        "temperature":250, 
        "pressure":30, 
        "runningState": true 
    } 
}

Configuration LKV :

Champ Value
Chemin d’accès entrant* .payload.temperature
Chemin de sortie .payload.lkvtemperature
Heure d’expiration 10h
Chemin d’accès du timestamp .payload.lkvtemperature_timestamp

Les valeurs LKV suivies sont les suivantes :

  • .payload.temperature est de 250.
  • Le timestamp de la valeur LKV est de 2023-01-11T10:02:07Z

Pour un message qui arrive à 11:05 avec une charge utile qui n’a pas la propriété de température, l’index LKV enrichit le message avec les valeurs suivies :

Exemple d’entrée à l’index LKV à 11:05 :

{ 
    "systemProperties":{ 
        "partitionKey":"pump", 
        "partitionId":5, 
        "timestamp":"2023-01-11T11:05:00Z" 
    }, 
    "qos":1, 
    "topic":"/assets/pump/#" 
    }, 
    "payload":{ 
        "runningState": true 
    } 
}

Exemple de sortie de l’index LKV à 11:05 :

{ 
    "systemProperties":{ 
        "partitionKey":"pump", 
        "partitionId":5, 
        "timestamp":"2023-01-11T11:05:00Z" 
    }, 
    "qos":1, 
    "topic":"/assets/pump/#" 
    }, 
    "payload":{ 
        "lkvtemperature":250, 
        "lkvtemperature_timestamp"":"2023-01-11T10:02:07Z" 
        "runningState": true 
    } 
}

Contenu connexe