Tutoriel : Exécuter Azure Functions à partir des travaux Azure Stream Analytics

Article
04/02/2024

Dans ce tutoriel, vous créez un travail Azure Stream Analytics qui lit des événements dans Azure Event Hubs, exécute une requête sur les données d’événement, puis appelle une fonction Azure qui écrit dans une instance Azure Cache pour Redis.

Capture d’écran montrant la relation entre les services Azure dans la solution.

Remarque

Vous pouvez exécuter Azure Functions à partir d’Azure Stream Analytics en configurant Functions comme l’un des récepteurs (de sortie) de la tâche Stream Analytics. Functions offre une expérience de calcul à la demande basée sur des événements, qui vous permet d’implémenter le code qui est déclenché par les événements qui se produisent dans Azure ou dans des services tiers. Comme ce logiciel peut répondre à des déclencheurs, il constitue l’outil de sortie logique pour des travaux Stream Analytics.
Stream Analytics appelle Functions via des déclencheurs HTTP. L’adaptateur de sortie de Functions permet aux utilisateurs de connecter Functions à Stream Analytics, de telle sorte que les événements puissent être déclenchés en fonction des requêtes Stream Analytics.
Il n’est pas possible d’établir une connexion à Azure Functions au sein d’un réseau virtuel à partir d’une tâche Stream Analytics qui s’exécute dans un cluster multilocataire.

Dans ce tutoriel, vous allez apprendre à :

Créer une instance Azure Event Hubs
Créer une instance Cache Redis Azure
Création d’une fonction Azure
Création d’un travail Stream Analytics
Configurer un hub d’événements comme entrée et une fonction comme sortie
Exécuter la tâche Stream Analytics
Consulter les résultats dans Cache Redis Azure

Si vous n’avez pas d’abonnement Azure, créez un compte gratuit avant de commencer.

Prérequis

Avant de commencer, vous devez avoir suivi les étapes ci-dessous :

Si vous n’avez pas d’abonnement Azure, créez un compte gratuit.
Téléchargez l’application de génération d’événements d’appel téléphonique, TelcoGenerator.zip à partir du Centre de téléchargement Microsoft, ou obtenez le code source à partir de GitHub.

Connectez-vous au portail Azure.

Créer un hub d’événements

Vous devez envoyer des exemples de données à un hub d’événements avant l’analyse par Stream Analytics du flux de données d’appels frauduleux. Dans ce didacticiel, vous envoyez des données à Azure à l’aide d’Azure Event Hubs.

Suivez les étapes ci-dessous pour créer un hub d’événements et envoyer les données d’appel à celui-ci :

Connectez-vous au portail Azure.
Sélectionnez Tous les services dans le menu de gauche, sélectionnez Internet des objets, pointez le curseur de la souris sur Event Hubs, puis sélectionnez le bouton + (Ajouter).
Dans la page Créer un espace de noms, suivez ces étapes :
1. Sélectionnez un abonnement Azure dans lequel vous souhaitez créer le hub d’événements.
2. Pour Groupe de ressources, sélectionnez Créer un nouveau et entrez un nom pour le groupe de ressources. L’espace de noms Event Hubs est créé dans ce groupe de ressources.
3. Pour le Nom de l’espace de noms, entrez un nom unique pour l’espace de noms Event Hubs.
4. Pour Emplacement, sélectionnez la région dans laquelle vous souhaitez créer l’espace de noms.
5. Pour Niveau tarifaire, sélectionnez Standard.
6. Au bas de la page, sélectionnez Examiner et créer.
7. Dans la page Vérifier et créer de l’Assistant Création d’espace de noms, sélectionnez Créer en bas de la page après avoir examiné tous les paramètres.
Une fois l’espace de noms correctement déployé, sélectionnez Accéder à la ressource pour aller à la page Espace de noms Event Hubs.
Dans la page Espace de noms Event Hubs, sélectionnez +Event Hub dans la barre de commandes.
Dans la page Créer un Event Hub, entrez un Nom pour le hub d’événements. Affectez la valeur 2 au Nombre de partitions. Utilisez les options par défaut pour les autres paramètres, puis sélectionnez Vérifier et créer.
En bas de la page Vérifier + créer, sélectionnez Créer. Ensuite, attendez que le déploiement se termine.

Accorder l’accès au concentrateur Event Hub et obtenir une chaîne de connexion

Pour qu’une application puisse envoyer des données à Azure Event Hubs, le hub d’événements doit disposer d’une stratégie autorisant l’accès. La stratégie d’accès génère une chaîne de connexion qui inclut des informations d’autorisation.

Dans la page Espace de noms Event Hubs, sélectionnez Stratégies d’accès partagé dans le menu de gauche.
Sélectionnez RootManageSharedAccessKey dans la liste des stratégies.
Ensuite, sélectionnez le bouton Copier à côté de Chaîne de connexion – Clé primaire.
Collez la chaîne de connexion dans un éditeur de texte. Vous avez besoin de cette chaîne de connexion dans la section suivante.

La chaîne de connexion ressemble à ceci :

Endpoint=sb://<Your event hub namespace>.servicebus.windows.net/;SharedAccessKeyName=<Your shared access policy name>;SharedAccessKey=<generated key>

Notez que la chaîne de connexion contient plusieurs paires clé-valeur séparées par des points-virgules : Endpoint, SharedAccessKeyName et SharedAccessKey.

Démarrer l’application de génération d’événements

Avant de démarrer l’application TelcoGenerator, vous devez la configurer pour envoyer des données aux Azure Event Hubs que vous avez créés précédemment.

Extrayez le contenu du fichier TelcoGenerator.zip.
Ouvrez le fichier TelcoGenerator\TelcoGenerator\telcodatagen.exe.config dans l’éditeur de texte de votre choix (comme il existe plusieurs fichiers .config, veillez à ouvrir celui qui convient).
Mettez à jour l’élément <appSettings> dans le fichier de configuration avec les détails suivants :
- Définissez la valeur de la clé EventHubName sur la valeur d’EntityPath à la fin de la chaîne de connexion.
- Définissez la valeur de la clé Microsoft.ServiceBus.ConnectionString sur la chaîne de connexion à l’espace de noms. Si vous utilisez une chaîne de connexion à un hub d’événements et non pas à un espace de noms, supprimez la valeur EntityPath (;EntityPath=myeventhub) à la fin. N’oubliez pas de supprimer le point-virgule qui précède la valeur d’EntityPath.
Enregistrez le fichier .

Ensuite, ouvrez une fenêtre de commandes et accédez au dossier dans lequel l’application TelcoGenerator est décompressée. Puis, entrez la commande suivante :

.\telcodatagen.exe 1000 0.2 2

Cette commande utilise les paramètres suivants :

Nombre d’enregistrements de données d’appel par heure.
Pourcentage de probabilité de fraude, qui correspond à la fréquence à laquelle l’application doit simuler un appel frauduleux. La valeur 0,2 signifie qu’environ 20 % des enregistrements d’appels semblent frauduleux.
Durée en heures, qui correspond au nombre d’heures pendant lesquelles l’application doit s’exécuter. Vous pouvez également arrêter l’application à tout moment en terminant le processus (Ctrl+C) sur la ligne de commande.

Après quelques secondes, l’application commence à afficher des enregistrements des appels téléphoniques à l’écran à mesure qu’elle les envoie au concentrateur Event Hub. Les données d’appel téléphonique contiennent les champs suivants :

Enregistrement	Définition
CallrecTime	Horodatage de l’heure de début d’appel.
SwitchNum	Commutateur téléphonique utilisé pour connecter l’appel. Pour cet exemple, les commutateurs sont des chaînes qui représentent le pays/la région d’origine (États-Unis, Chine, Royaume-Uni, Allemagne ou Australie).
CallingNum	Numéro de téléphone de l’appelant.
CallingIMSI	Identité de l’abonné mobile international (IMSI). Il s’agit d’un identificateur unique de l’appelant.
CalledNum	Numéro de téléphone du destinataire de l’appel.
CalledIMSI	Identité de l'abonné mobile international (IMSI). Il s’agit d’un identificateur unique du destinataire de l’appel.

Création d’un travail Stream Analytics

Maintenant que vous disposez d’un flux d’événements d’appel, vous pouvez créer un travail Stream Analytics qui lit des données à partir du hub d’événements.

Pour créer un travail Stream Analytics, accédez au portail Azure.
Sélectionnez Créer une ressource, puis recherchez Tâche Stream Analytics. Sélectionnez la vignette Tâche Stream Analytics et sélectionnez Créer.
Sur la page Nouveau travail Stream Analytics, procédez comme suit :
1. Pour Abonnement, sélectionnez l’abonnement qui contient l’espace de noms Event Hubs.
2. Pour Groupe de ressources, sélectionnez le groupe de ressources que vous avez créé précédemment.
3. Dans la section Détails de l’instance , pour Nom, entrez un nom unique pour le travail Stream Analytics.
4. Pour Région, sélectionnez la région dans laquelle vous souhaitez créer le travail Stream Analytics. Nous vous recommandons de placer le travail et le concentrateur Event Hub dans la même région afin d’optimiser les performances. Ce faisant, vous ne payez pas pour transférer des données entre les régions.
5. Pour Environnement d’hébergement<, sélectionnez Cloud s’il n’est pas déjà sélectionné. Les travaux Stream Analytics peuvent être déployés dans le cloud ou sur des appareils Edge. L’option Cloud vous permet de déployer votre travail dans le cloud Azure, et l’option Edge sur un appareil IoT Edge.
6. Dans le champ Unités de streaming, sélectionnez 1. Les unités de streaming sont les ressources de calcul requises pour exécuter un travail. Par défaut, cette valeur est définie sur 1. Pour en savoir plus sur la mise à l’échelle des unités de streaming, consultez Understanding and adjusting streaming units (Présentation et réglage des unités de streaming).
7. Au bas de la page, sélectionnez Examiner et créer.
Sur la page Examiner et créer, examinez les paramètres, puis sélectionnez Créer pour créer une tâche Stream Analytics.
Une fois le travail déployé, sélectionnez Accéder à la ressource pour accéder à la page Tâche Stream Analytics.

Configurer les entrées du travail

L’étape suivante consiste à définir une source d’entrée pour le travail, afin de pouvoir lire les données à l’aide de l’Event Hub que vous avez créé dans la section précédente.

Sur la page Tâche Stream Analytics, dans la section Topologie de la tâche dans le menu de gauche, sélectionnez Entrées.
Dans la page Entrées, sélectionnez + Ajouter une entrée et Event Hub.
Sur la page Event Hub, procédez comme suit :
1. Pour Alias d’entrée, entrez CallStream. L’Alias d’entrée est un nom convivial pour identifier votre entrée. L’alias d’entrée peut contenir uniquement des caractères alphanumériques, des traits d’union et des traits de soulignement, et doit avoir entre 3 et 63 caractères.
2. Pour Abonnement, sélectionnez l’abonnement Azure dans lequel vous avez créé le Event Hub. Le hub d’événements peut se trouver dans le même abonnement ou dans un autre abonnement que le travail Stream Analytics.
3. Pour Espace de noms Event Hubs, sélectionnez l’espace de noms du Event Hubs que vous avez créé dans la section précédente. Tous les espaces de noms disponibles dans votre abonnement actuel sont répertoriés dans la liste déroulante.
4. Pour le nom Event Hub, sélectionnez l’Event Hub que vous avez créé dans la section précédente. Tous les Event Hubs disponibles dans l’espace de noms sont répertoriés dans la liste déroulante.
5. Pour le groupe de consommateurs Event Hub, conservez l’option Créer un nouveau sélectionnée afin qu’un nouveau groupe de consommateurs soit créé sur l’Event Hub. Nous vous recommandons d’utiliser un groupe de consommateurs différent pour chaque travail Stream Analytics. Si aucun groupe de consommateurs n’est spécifié, le travail Stream Analytics utilise le groupe de consommateurs $Default. Lorsqu’un travail contient une jointure réflexive ou plusieurs entrées, certaines entrées peuvent être lues ultérieurement par plusieurs lecteurs. Cette situation a une incidence sur le nombre de lecteurs dans un groupe de consommateurs unique.
6. Dans le champ Mode d’authentification, sélectionnez Chaîne de connexion. Il est plus facile de tester le tutoriel avec cette option.
7. Pour Nom de la stratégie du Event Hub, sélectionnez Utiliser existant, puis sélectionnez la stratégie que vous avez créée précédemment.
8. Sélectionnez Enregistrer au bas de la page.

Créer une instance Cache Redis Azure

Créez un cache dans Cache Redis Azure en suivant les étapes décrites dans la section Create a cache (Créer un cache).
Une fois le cache créé, sous Paramètres, sélectionnez Clés d’accès. Notez la chaîne de connexion primaire.

Créer une fonction dans Azure Functions qui permet d’écrire des données dans Cache Redis Azure

Consultez la section Créer une application de fonction de la documentation Functions. Cet exemple a été construit sur :
- Runtime Azure Functions version 4
- .NET 6.0
- StackExchange.Redis 2.2.8
Créez une application de fonction HttpTrigger par défaut dans Visual Studio Code en suivant ce didacticiel. Les informations suivantes sont utilisées : Lange : C#, Runtime : .NET 6 (sous la fonction v4), Modèle : HTTP trigger.
Installez la bibliothèque de client Redis en exécutant la commande suivante dans un terminal situé dans le dossier du projet :
```
dotnet add package StackExchange.Redis --version 2.2.88
```
Ajoutez les éléments RedisConnectionString et RedisDatabaseIndex dans la section Values de votre fichier local.settings.json, en renseignant la chaîne de connexion du serveur de destination :
```
{
    "IsEncrypted": false,
    "Values": {
        "AzureWebJobsStorage": "",
        "FUNCTIONS_WORKER_RUNTIME": "dotnet",
        "RedisConnectionString": "Your Redis Connection String",
        "RedisDatabaseIndex":"0"
    }
}
```
L’index de base de données Redis est le nombre de 0 à 15 identifiant la base de données sur l’instance.

Remplacez la fonction entière (fichier .cs dans le projet) par l’extrait de code suivant. Mettez à jour l’espace de noms, le nom de la classe et le nom de la fonction en les remplaçant par les vôtres :

using System;
using System.IO;
using System.Threading.Tasks;
using Microsoft.AspNetCore.Mvc;
using Microsoft.Azure.WebJobs;
using Microsoft.Azure.WebJobs.Extensions.Http;
using Microsoft.AspNetCore.Http;
using Microsoft.Extensions.Logging;
using Newtonsoft.Json;

using StackExchange.Redis;

namespace Company.Function
{
    public static class HttpTrigger1{
        [FunctionName("HttpTrigger1")]
        public static async Task<IActionResult> Run(
            [HttpTrigger(AuthorizationLevel.Function, "get","post", Route = null)] HttpRequest req,
            ILogger log)
        {
            // Extract the body from the request
            string requestBody = await new StreamReader(req.Body).ReadToEndAsync();
            if (string.IsNullOrEmpty(requestBody)) {return new StatusCodeResult(204);} // 204, ASA connectivity check

            dynamic data = JsonConvert.DeserializeObject(requestBody);

            // Reject if too large, as per the doc
            if (data.ToString().Length > 262144) {return new StatusCodeResult(413);} //HttpStatusCode.RequestEntityTooLarge

            string RedisConnectionString = Environment.GetEnvironmentVariable("RedisConnectionString");
            int RedisDatabaseIndex = int.Parse(Environment.GetEnvironmentVariable("RedisDatabaseIndex"));

            using (var connection = ConnectionMultiplexer.Connect(RedisConnectionString))
            {
                // Connection refers to a property that returns a ConnectionMultiplexer
                IDatabase db = connection.GetDatabase(RedisDatabaseIndex);

                // Parse items and send to binding
                for (var i = 0; i < data.Count; i++)
                {
                    string key = data[i].Time + " - " + data[i].CallingNum1;

                    db.StringSet(key, data[i].ToString());
                    log.LogInformation($"Object put in database. Key is {key} and value is {data[i].ToString()}");

                    // Simple get of data types from the cache
                    string value = db.StringGet(key);
                    log.LogInformation($"Database got: {key} => {value}");

                }
            }
            return new OkResult(); // 200
        }
    }
}

Quand Stream Analytics reçoit de la fonction l’exception « HTTP Request Entity Too Large » (Entité de requête HTTP trop volumineuse), il réduit la taille des lots envoyés aux fonctions. Le code suivant garantit que Stream Analytics n’envoie pas de lots trop volumineux. Vérifiez également que les valeurs de taille et de nombre de lots maximum utilisées dans la fonction correspondent à celles qui ont été saisies dans le portail Stream Analytics.

La fonction peut désormais être publiée sur Azure.
Ouvrez la fonction sur le Portail Azure et définissez les paramètres d’application pour RedisConnectionString et RedisDatabaseIndex .

Mettre à jour le travail Stream Analytics en utilisant la fonction comme sortie

Dans le portail Azure, ouvrez votre travail Stream Analytics.

Accédez à votre fonction et sélectionnez Vue d’ensemble>Sorties>Ajouter. Pour ajouter une nouvelle sortie, sélectionnez Azure Function pour l’option de récepteur. L’adaptateur de sortie Functions a les propriétés suivantes :

Nom de la propriété	Description
Alias de sortie	Nom convivial que vous utilisez dans la requête du travail pour faire référence à la sortie.
Option d’importation	Vous pouvez utiliser la fonction de l’abonnement actuel ou renseigner manuellement les paramètres si la fonction se trouve dans un autre abonnement.
Function App	Nom de votre application Functions.
Fonction	Nom de la fonction dans votre application Functions (nom de votre fonction run.csx).
Taille de lot maximale	Définit la taille maximale de chaque lot de sortie envoyé à votre fonction, en octets. Par défaut, cette valeur est définie à 262 144 octets (256 Ko).
Nombre maximal de lots	Spécifie le nombre maximal d’événements dans chaque lot envoyé à la fonction. La valeur par défaut est 100. Cette propriété est facultative.
Clé	Vous permet d’utiliser une fonction d’un autre abonnement. Indiquez la valeur de la clé pour accéder à votre fonction. Cette propriété est facultative.

Indiquez un nom pour l’alias de sortie. Dans ce didacticiel, il est nommé saop1, mais vous pouvez utiliser n’importe quel nom de votre choix. Renseignez les autres détails.

Ouvrez votre travail Stream Analytics et mettez à jour la requête comme suit.

Important

L’exemple de script suivant suppose que vous avez utilisé CallStream pour le nom de l’entrée et saop1 pour le nom de la sortie. Si vous avez utilisé des noms différents, n’oubliez PAS de mettre à jour la requête.

 SELECT
         System.Timestamp as Time, CS1.CallingIMSI, CS1.CallingNum as CallingNum1,
         CS2.CallingNum as CallingNum2, CS1.SwitchNum as Switch1, CS2.SwitchNum as Switch2
     INTO saop1
     FROM CallStream CS1 TIMESTAMP BY CallRecTime
        JOIN CallStream CS2 TIMESTAMP BY CallRecTime
         ON CS1.CallingIMSI = CS2.CallingIMSI AND DATEDIFF(ss, CS1, CS2) BETWEEN 1 AND 5
     WHERE CS1.SwitchNum != CS2.SwitchNum

Démarrez l’application telcodatagen.exe en exécutant la commande suivante dans la ligne de commande. La commande utilise le format telcodatagen.exe [#NumCDRsPerHour] [SIM Card Fraud Probability] [#DurationHours].
```
telcodatagen.exe 1000 0.2 2
```
Démarrez le travail Stream Analytics.
Dans la page Surveiller votre fonction Azure, vous voyez que la fonction est appelée.
Dans la page Azure Cache pour Redis votre cache, sélectionnez Métriques dans le menu de gauche, ajoutez une métrique d’écriture du cache et définissez la durée sur la dernière heure. Le résultat ressemble à l’image suivante :

Consulter les résultats dans Cache Redis Azure

Obtenir la clé à partir des journaux Azure Functions

Tout d’abord, obtenez la clé d’un enregistrement inséré dans Azure Cache pour Redis. Dans le code, la clé est calculée dans la fonction Azure, comme indiqué dans l’extrait de code suivant :

string key = data[i].Time + " - " + data[i].CallingNum1;

db.StringSet(key, data[i].ToString());
log.LogInformation($"Object put in database. Key is {key} and value is {data[i].ToString()}");

Accédez au Portail Azure et recherchez votre application Azure Functions.
Sélectionnez Fonctions dans le menu de gauche.
Sélectionnez HTTPTrigger1 dans la liste des fonctions.
Sélectionnez Surveiller dans le menu de gauche.
Basculez vers l’onglet Journaux.
Notez une clé du message d’information, comme illustré dans la capture d’écran suivante. Vous utilisez cette clé pour rechercher la valeur dans Azure Cache pour Redis.

Utilisez la clé pour rechercher l’enregistrement dans Azure Cache pour Redis

Accédez au Portail Azure et recherchez votre instance Cache Redis Azure. Sélectionnez Console.
Utilisez les commandes Cache Redis Azure pour vérifier que vos données se trouvent dans Cache Redis Azure. (La commande prend le format Get {key}.) Utilisez la clé que vous avez copiée à partir des journaux Monitor pour la fonction Azure (dans la section précédente).

Get "KEY-FROM-THE-PREVIOUS-SECTION"

Cette commande doit produire la valeur de la clé spécifiée :

Gestion des erreurs et nouvelles tentatives

Si une défaillance se produit lors de l’envoi d’événements à Azure Functions, Stream Analytics retente la plupart des opérations. Toutes les exceptions HTTP sont retentées jusqu’à réussite, à l’exception de l’erreur HTTP 413 (Entité trop volumineuse). L’erreur causée par une entité trop volumineuse est traitée comme une erreur de données soumise à la stratégie Retenter ou supprimer.

Notes

Le délai d’expiration des requêtes HTTP adressées à Azure Functions par Stream Analytics est défini sur 100 secondes. Si le traitement d’un lot par votre application Azure Functions prend plus de 100 secondes, Stream Analytics génère une erreur et retente l’opération.

Toute nouvelle tentative après l’expiration du délai peut entraîner l’écriture d’événements en doublon dans le récepteur de sortie. Lorsque Stream Analytics retente l’opération pour un lot ayant échoué, la nouvelle tentative concerne tous les événements dans le lot. Prenons l’exemple d’un lot de 20 événements qui sont envoyés à Azure Functions à partir de Stream Analytics. Supposons qu’Azure Functions traite les 10 premiers événements de ce lot en 100 secondes. Après 100 secondes, Stream Analytics interrompt la requête, car il n’a pas reçu de réponse positive de la part d’Azure Functions, et une autre requête est envoyée pour le même lot. Les dix premiers événements du lot sont retraités par Azure Functions, ce qui crée un doublon.

Problèmes connus

Dans le portail Azure, lorsque vous essayez de rétablir les valeurs maximales de taille et de nombre de lots à des valeurs vides (par défaut), la valeur bascule sur la valeur précédemment saisie au moment de la sauvegarde. Dans ce cas, entrez manuellement les valeurs par défaut pour ces champs.

Actuellement, l’utilisation du routage HTTP sur votre instance Azure Functions n’est pas prise en charge par Stream Analytics.

La prise en charge de la connexion aux applications Azure Functions hébergées dans un réseau virtuel n’est pas activée.

Nettoyer les ressources

Lorsque vous n’en avez plus besoin, supprimez le groupe de ressources, le travail de streaming et toutes les ressources associées. La suppression du travail évite la facturation des unités de streaming consommées par le travail. Si vous envisagez d’utiliser le travail à l’avenir, vous pouvez l’arrêter et le redémarrer plus tard lorsque vous en avez besoin. Si vous ne comptez pas continuer à utiliser ce travail, supprimez toutes les ressources créées dans le cadre de ce guide de démarrage rapide en procédant comme suit :

Dans le menu de gauche du portail Azure, cliquez sur Groupes de ressources, puis sur le nom de la ressource que vous avez créée.
Sur la page de votre groupe de ressources, sélectionnez Supprimer, saisissez le nom de la ressource à supprimer dans la zone de texte, puis sélectionnez Supprimer.

Étapes suivantes

Dans ce tutoriel, vous avez créé une tâche Stream Analytics simple qui exécute une fonction Azure. Pour en savoir plus sur les travaux Stream Analytics, passez au didacticiel suivant :

Mettre à jour ou fusionner des enregistrements dans Azure SQL Database avec Azure Functions Exécuter des fonctions JavaScript définies par l’utilisateur dans des travaux Stream Analytics