Utiliser des données de référence pour effectuer des recherches dans Stream Analytics

Article
02/26/2024

Les données de référence sont un jeu de données fini statique ou à variation lente par nature. Elles sont utilisées pour effectuer une recherche ou pour augmenter vos flux de données. Les données de référence sont également appelées tables de recherche.

Prenons un exemple de scénario IoT. Vous pouvez stocker les métadonnées relatives aux capteurs, qui ne changent pas souvent, dans les données de référence. Vous pouvez alors la joindre aux flux de données IoT en temps réel.

Azure Stream Analytics charge les données de référence dans la mémoire pour obtenir un traitement de flux à faible latence. Pour utiliser des données de référence dans votre tâche Stream Analytics, vous utiliserez généralement une jointure de données de référence dans votre requête.

Exemple

Vous pouvez avoir un flux d’événements en temps réel généré quand les voitures passent un poste de péage. Le guichet peut capturer les plaques de licence en temps réel. Ces données peuvent être jointes à un jeu de données statique qui possède des détails d’inscription pour identifier les plaques de licence qui ont expiré.

SELECT I1.EntryTime, I1.LicensePlate, I1.TollId, R.RegistrationId  
FROM Input1 I1 TIMESTAMP BY EntryTime  
JOIN Registration R  
ON I1.LicensePlate = R.LicensePlate  
WHERE R.Expired = '1'

Stream Analytics prend en charge le stockage Blob Azure, Azure Data Lake Storage Gen2 et Azure SQL Database comme couche de stockage pour les données de référence. Si les données de référence se trouvent dans d’autres magasins de données, essayez d’utiliser Azure Data Factory pour extraire, transformer et charger les données dans l’un des magasins de données pris en charge. Pour plus d’informations, consultez Vue d’ensemble de l’activité de copie dans Azure Data Factory.

Stockage Blob Azure ou Azure Data Lake Storage Gen2

Les données de référence sont modélisées en tant que séquence d'objets Blob dans l'ordre croissant de la date/l'heure spécifiée dans le nom de l'objet blob. Les Blobs peuvent être ajoutés à la fin de la séquence à l’aide d’une date/heure ultérieure à celle indiquée par le dernier objet blob dans la séquence. Les objets Blob sont définis dans la configuration d’entrée.

Pour plus d’informations, consultez Utiliser les données de référence d’un Stockage Blob pour un travail Stream Analytics.

Configurer les données de référence d’objet blob

Pour configurer vos données de référence, vous devez d'abord créer une entrée de type Données de référence. Le tableau ci-dessous explique chaque propriété que vous devez fournir lors de la création de l’entrée des données de référence avec sa description.

Nom de la propriété	Description
Alias d’entrée	Nom convivial qui servira dans la requête de tâche pour faire référence à cette entrée.
Compte de stockage	Nom du compte de stockage dans lequel se trouvent vos blobs. S’il se trouve dans le même abonnement que votre travail Stream Analytics, vous pouvez le sélectionner dans la liste déroulante.
Clé du compte de stockage	Clé secrète associée au compte de stockage. Cette clé est remplie automatiquement si le compte de stockage se trouve dans le même abonnement que votre travail Stream Analytics.
Conteneur de stockage	Les conteneurs fournissent un regroupement logique des objets blob stockés dans le stockage Blob. Lorsque vous téléchargez un objet blob dans le stockage d'objets Blob, vous devez spécifier un conteneur pour cet objet blob.
Modèle de chemin d’accès	C’est une propriété obligatoire qui est utilisée pour localiser vos objets blob dans le conteneur spécifié. Dans le chemin d'accès, vous pouvez choisir de spécifier une ou plusieurs instances de l'une des variables : {date} et {time}. Exemple 1 : products/{date}/{time}/product-list.csv Exemple 2 : products/{date}/product-list.csv Exemple 3 : product-list.csv Si l’objet blob n’existe pas dans le chemin d’accès spécifié, le travail Stream Analytics attend indéfiniment que l’objet blob devienne disponible.
Format de date [facultatif]	Si vous avez utilisé {date} dans le modèle de chemin d’accès que vous avez spécifié, sélectionnez le format de date selon lequel vos blobs sont organisés dans la liste déroulante des formats pris en charge. Exemple : AAAA/MM/JJ, MM/JJ/AAAA
Format de l’heure [facultatif]	Si vous avez utilisé {time} dans le modèle de chemin d’accès que vous avez spécifié, sélectionnez le format d’heure selon lequel vos blobs sont organisés dans la liste déroulante des formats pris en charge. Exemple : HH, HH/mm ou HH-mm
Format de sérialisation de l’événement	Pour s’assurer que vos requêtes fonctionnent comme prévu, Stream Analytics a besoin de connaître le format de sérialisation que vous utilisez pour les flux de données d’entrée. Pour les données de référence, les formats pris en charge sont CSV et JSON.
Encodage	UTF-8 est le seul format de codage actuellement pris en charge.

Données de référence statiques

Vos données de référence peuvent ne pas être susceptibles d’être modifiées. Pour activer la prise en charge des données de référence statiques, spécifiez un chemin d’accès statique dans la configuration d’entrée.

Stream Analytics récupère l’objet blob à partir du chemin spécifié. Les jetons de substitution {date} et {time} ne sont pas nécessaires. Étant donné que les données de référence sont immuables dans Stream Analytics, le remplacement d’un objet blob de données de référence statiques n’est pas recommandé.

Générer des données de référence dans une planification

Vos données de référence peuvent être un jeu de données à variation lente. Pour rafraîchir les données de référence, spécifiez un modèle de chemin dans la configuration d'entrée en utilisant les jetons de substitution {date} et {time}. Stream Analytics collectera les définitions de données de référence mises à jour en fonction de ce modèle de chemin d’accès.

Par exemple, un modèle sample/{date}/{time}/products.csv avec un format de date « AAAA-MM-JJ » et un format d’heure « HH:mm » donne pour instruction à Stream Analytics de sélectionner l’objet blob sample/2015-04-16/17-30/products.csv mis à jour à 17:30 le 16 avril 2015 (UTC).

Stream Analytics analyse automatiquement les objets blob de données de référence actualisées à un intervalle d’une minute. Un objet blob avec l’horodateur 10:30:00 peut être téléchargé avec un léger délai, par exemple 10:30:30. Vous remarquez un petit retard dans le travail de Stream Analytics qui référence cet objet blob.

Pour éviter de tels scénarios, chargez l’objet blob antérieur à l’heure d’effet cible, qui est 10:30:00 dans cet exemple. La tâche de Stream Analytics a maintenant suffisamment de temps pour découvrir et charger l’objet blob en mémoire et effectuer des opérations.

Notes

Actuellement, les tâches Stream Analytics recherchent l’actualisation des objets blob uniquement lorsque l’heure machine s’approche de l’heure encodée dans le nom de l’objet blob. Par exemple, la tâche recherchée pour sample/2015-04-16/17-30/products.csv est le plus tôt possible, mais pas avant le 16 avril 2015, à 17:30 UTC. Elle ne recherche jamais un blob avec une heure encodée antérieure à la dernière heure détectée.

Par exemple, une fois que le travail a trouvé l’objet blob sample/2015-04-16/17-30/products.csv, il ignore tous les fichiers ayant une date encodée antérieure au 16 avril 2015, à 17:30. Si un objet blob sample/2015-04-16/17-25/products.csv à l’arrivée tardive est créé dans le même conteneur, la tâche ne l’utilise pas.

Dans un autre exemple, sample/2015-04-16/17-30/products.csv est produit le 16 avril 2015, à 10:03, mais aucun objet blob avec une date antérieure n’est présent dans le conteneur. Ensuite, la tâche utilise ce fichier à partir du 16 avril 2015 à 10:03 et utilise les données de référence précédentes jusqu’à ce moment-là.

La seule exception à ce comportement est quand le travail doit retraiter des données antérieures ou quand il est démarré pour la première fois.

Au moment du démarrage, la tâche recherche l'objet blob le plus récent produit avant l'heure de début de la tâche spécifiée. Ce comportement garantit qu’il y a un jeu de données de référence non vide au démarrage du travail. Si un aucun jeu de données n’est trouvé, la tâche affiche le diagnostic suivant : Initializing input without a valid reference data blob for UTC time <start time>.

Lorsqu’un jeu de données de référence est actualisé, un journal de diagnostic est généré : Loaded new reference data from <blob path>. Pour de nombreuses raisons, un travail peut avoir besoin de recharger un jeu de données de référence précédent. Le plus souvent, la raison est de retraiter les données passées. Le même journal de diagnostic est généré à ce moment-là. Cette action ne signifie pas que les données de flux actuelles utilisent des données de référence antérieures.

Azure Data Factory peut être utilisé pour orchestrer la tâche de création de blobs mis à jour requise par Stream Analytics pour mettre à jour les définitions de données de référence.

Data Factory est un service d’intégration de données dans le cloud qui gère et automatise le déplacement et la transformation des données. Data Factory prend en charge la connexion à un grand nombre de magasins de données cloud et en local. Il peut facilement déplacer des données selon une planification régulière que vous spécifiez.

Pour plus d'informations sur la façon de configurer un pipeline Data Factory pour générer des données de référence pour Stream Analytics qui est actualisé selon une planification prédéfinie, consultez cet exemple GitHub.

Conseils pour l'actualisation des données de référence d’objets blob

Ne remplacez pas les objets blob de données de référence, car ils sont immuables.
La méthode recommandée pour actualiser les données de référence consiste à :
- Utiliser {date}/{time} dans le modèle de chemin d’accès.
- Ajouter un nouvel objet blob à l’aide du même conteneur et du même chemin d’accès que ceux définis dans l’entrée du travail.
- Utiliser une date/heure supérieure à celle spécifiée par le dernier objet blob dans la séquence.
Les objets blob de données de référence ne sont pas* triés en fonction de l’heure de Dernière modification de l’objet blob. Elles sont classées uniquement selon la date et l’heure spécifiées dans le nom de l’objet blob à l’aide des substitutions {date} et {time}.
Afin d’éviter de se retrouver avec une liste de blobs volumineuse, supprimez les blobs très anciens pour lesquels le traitement ne se fera plus. Stream Analytics peut avoir à retraiter une petite quantité dans certains scénarios, comme un redémarrage.

Azure SQL Database

Votre travail de Stream Analytics récupère les données de référence SQL Database et les stocke sous la forme d’un instantané en mémoire à des fins de traitement. L’instantané de vos données de référence est également stocké dans un conteneur dans un compte de stockage. Vous spécifiez le compte de stockage dans les paramètres de configuration.

Le conteneur est créé automatiquement lorsque le travail démarre. Si le travail est arrêté ou entre dans un état d’échec, les conteneurs créés automatiquement sont supprimés une fois le travail redémarré.

Si vos données de référence sont un jeu de données à variation lente, vous devez régulièrement actualiser l’instantané qui est utilisé dans votre tâche.

Avec Stream Analytics vous permet de définir une fréquence de rafraîchissement lorsque vous configurez votre connexion d’entrée SQL Database. Le runtime Stream Analytics va interroger votre instance SQL Database à l’intervalle spécifié par la fréquence d’actualisation. La fréquence de rafraîchissement la plus rapide prise en charge correspond à une fois par minute. Pour chaque actualisation, Stream Analytics stocke un nouvel instantané dans le compte de stockage fourni.

Stream Analytics fournit deux options pour l’interrogation de votre instance SQL Database. Une requête de capture instantanée est obligatoire et doit être incluse dans chaque tâche. Stream Analytics exécute régulièrement la requête d’instantané en fonction de votre intervalle d’actualisation. Elle utilise le résultat de la requête (l’instantané) comme jeu de données de référence.

La requête d’instantané doit s’adapter à la plupart des scénarios. Si vous rencontrez des problèmes de performance avec de grands ensembles de données et des taux de rafraîchissement rapides, utilisez l'option de requête delta. Les requêtes qui prennent plus de 60 secondes pour retourner le jeu de données de référence se traduisent par un délai d’attente.

Avec l’option de requête delta, Stream Analytics exécute la requête de capture instantanée initialement pour obtenir un jeu de données de référence de ligne de base. Ensuite, Stream Analytics exécute la requête delta régulièrement selon votre intervalle d’actualisation pour récupérer les modifications incrémentielles. Ces modifications incrémentielles sont appliquées en permanence au jeu de données de référence pour le maintenir à jour. L’utilisation de requêtes delta peut permettre de réduire les coûts de stockage et les opérations d’e/s réseau.

Configurer les données de référence SQL Database

Pour configurer vos données de référence SQL Database, vous devez d'abord créer une entrée de données de référence. Le tableau ci-dessous explique chaque propriété que vous devez fournir lors de la création de l’entrée des données de référence avec sa description. Pour plus d'informations, consultez Utiliser les données de référence d'une instance de SQL Database pour une tâche Stream Analytics.

Vous pouvez utiliser Azure SQL Managed Instance comme entrée de données de référence. Vous pouvez configurer un point de terminaison public dans SQL Managed Instance. Ensuite, vous configurez manuellement les paramètres suivants dans Stream Analytics. Une machine virtuelle Azure exécutant SQL Server auquel une base de données est attachée est également prise en charge en configurant manuellement les paramètres ci-dessous.

Nom de la propriété	Description
Alias d’entrée	Nom convivial qui servira dans la requête de tâche pour faire référence à cette entrée.
Abonnement	Votre abonnement.
Base de données	Instance SQL Database qui contient vos données de référence. Pour SQL Managed Instance, vous devez spécifier le port 3342. par exemple `sampleserver.public.database.windows.net,3342`.
Nom d’utilisateur	Nom d’utilisateur associé à votre instance SQL Database.
Mot de passe	Mot de passe associé à votre instance SQL Database.
Actualiser régulièrement	Cette option vous permet de choisir une fréquence d’actualisation. Sélectionnez Activé pour spécifier la fréquence d’actualisation au format DD:HH:MM.
Requête d'instantané	Il s’agit de l’option de requête par défaut qui récupère les données de référence à partir de votre instance SQL Database.
Requête delta	Pour les scénarios avancés avec des jeux de données volumineux et une courte fréquence d’actualisation, choisissez d’ajouter une requête delta.

Limite de taille

Utilisez des jeux de données de référence qui sont inférieurs à 300 Mo pour des performances optimales. Les jeux de données de référence de 5 Go ou moins sont pris en charge dans les travaux avec six unités de diffusion en continu (SU) ou plus. L’utilisation d’un jeu de données de référence volumineux peut affecter la latence de bout en bout de votre travail.

La complexité des requêtes peut augmenter pour inclure le traitement avec état comme les agrégats à fenêtres, les jointures temporelles et les fonctions analytiques temporelles. Lorsque la complexité augmente, la taille maximale prise en charge des données de référence diminue.

Si Stream Analytics ne peut pas charger les données de référence et effectuer des opérations complexes, le travail manque de mémoire et échoue. Dans ce cas, la métrique d’utilisation de l’unité de streaming atteint 100 %.

Nombre d’unités de streaming	Taille recommandée
1	50 Mo ou moins
3	150 Mo ou moins
6 et au-delà	5 Go ou moins

La prise en charge de la compression n’est pas disponible pour les données de référence. Pour les jeux de données de référence d’une taille supérieure à 300 Mo, utilisez SQL Database comme option source avec requête delta pour obtenir des performances optimales. Si l’option de requête delta n’est pas utilisée dans de tels scénarios, vous voyez des pics dans la métrique de retard en filigrane chaque fois que l’ensemble de données de référence est actualisé.

Jointure de plusieurs jeux de données de référence dans un travail

Vous pouvez uniquement joindre une entrée de données de référence à une entrée de streaming. Par conséquent, pour joindre plusieurs jeux de données de référence, fractionnez la requête en plusieurs étapes. Voici un exemple :

With Step1 as (
    --JOIN input stream with reference data to get 'Desc'
    SELECT streamInput.*, refData1.Desc as Desc
    FROM    streamInput
    JOIN    refData1 ON refData1.key = streamInput.key 
)
--Now Join Step1 with second reference data
SELECT *
INTO    output 
FROM    Step1
JOIN    refData2 ON refData2.Desc = Step1.Desc

Travaux IoT Edge

Seules les données de référence locales sont prises en charge pour les travaux de périphérie Stream Analytics. Lorsqu'une tâche est déployée sur un appareil IoT Edge, il charge les données de référence à partir du chemin de fichier défini par l'utilisateur. Préparez un fichier de données de référence sur l'appareil.

Pour un conteneur Windows, placez le fichier de données de référence sur le lecteur local et partagez ce lecteur avec le conteneur Docker. Pour un conteneur Linux, créez un volume Docker et renseignez le fichier de données sur le volume.

Les données de référence situées sur une mise à jour d’IoT Edge sont déclenchées par un déploiement. Une fois déclenché, le module Stream Analytics sélectionne les données mises à jour sans arrêter la tâche en cours d’exécution.

Vous pouvez mettre à jour les données de référence de deux façons :

Mettre à jour le chemin des données de référence de votre tâche Stream Analytics à partir du portail Azure.
En mettant à jour le déploiement IoT Edge

Étapes suivantes

Démarrage rapide : Créer un travail Stream Analytics à l’aide du portail Azure

Partage via