Azure Event Hub & Stream Analytics–Part 2: analyse des évènements

Dans cette seconde partie, nous allons nous intéresser à l’analyse des évènements dans le nouveau service Stream Analytics de Microsoft Azure.

Nous nous attaquons donc à cette partie de notre architecture.

Le but de cette analyse va être d’agréger l’ensemble des appels d’API :

par scope
sur une période glissante de 1 min

Création du Hub d’évènement

Après avoir créer le middleware d’interception dans notre billet précédent, nous devons désormais créer le hub qui va recevoir nos évènements.

Nous nous rendons donc sur le portail pour créer un espace de nom de type service bus. Suivant le tableau de pricing/tiers du service bus, nous pouvons créer un compte de type basic pour profiter de l’event hub.

Nous nous rendons donc sur le portail pour créer un nouveau hub d’évènement, soit:

vous créer un hub d’évènement sur un espace de nom existant à l’aide de la création rapide
vous créer un hub d’évènement sur un nouvelle espace de nom à l’aide de la création personnalisée

La création personnalisée vous permet de gérer :

A l’étape 2, nous pouvons configurer le nombre de partitions ainsi que le nombre de jours de rétentions des messages :

Le minimum pour le nombre de partitions est de 8 et le maximum est de 32. (Au delà, vous devez appelez le support)
Le nombre de partition définit finalement votre degré de parallélisme requis vis à vis de votre application qui va consommer vos évènements

Mais c’est quoi une partition (msdn):

une partition est une séquence ordonnée d’évènements qui sont stockés dans l’event hub.

Lorsque l’on publie dans le hub, nous définissons une clé de partition, permettant notamment au niveau du publieur de s’assurer de publier des messages en FIFO.

La lecture des évènements dans les partitions peut se voir comme un curseur de lecture :

Il est de la responsabilité du souscripteur de savoir a quel offset il s’est arrêté dans le cas ou le programme de souscription s’arrête pour redémarrer au même endroit. C’est ici que la durée de rétention des messages joue alors sont rôles.

Nous voyons que contrairement au Service Bus + Topic, les messages sont stockées pendant une durée maximum et chaque consommateur se connecte a l’event hub pour retrouver les informations sur la période qui l’intéresse. (Dans le service bus, si aucune souscription n’est crée lorsqu’un message est publié dans un topic, alors le message est perdu).

Après la configuration de cet EventHub, nous pouvons mettre à jour nos paramètres de configuration de l’étape 1 et créer la base de données.

Création de la base de données

Nous allons créer une table dans une base de données SQL Azure pour pouvoir stocker les statistiques d’utilisation de nos APIs sur 5 minutes glissantes.

Voici la requêtes de création de la table :

 

CREATE TABLE [dbo].[ApiStatsByMin](
	[WinStartTime] [datetime2](6) NULL,
	[WinEndTime] [datetime2](6) NULL,
	[Scope] [nvarchar](50) NULL,
	[TotalRequestLength] [float] NULL,
	[TotalResponseLength] [float] NULL,
	[ApiCountCall] [bigint] NULL
)

/****** Object:  Index [AvgStatByMin]    Script Date: 20/11/2014 19:38:41 ******/
CREATE CLUSTERED INDEX ApiStatByMin ON [dbo].[ApiStatsByMin]
(
	[Scope] ASC
)WITH (PAD_INDEX = OFF, STATISTICS_NORECOMPUTE = OFF, SORT_IN_TEMPDB = OFF, DROP_EXISTING = OFF, ONLINE = OFF, ALLOW_ROW_LOCKS = ON, ALLOW_PAGE_LOCKS = ON)

Nous avons défini notre entrant pour notre Job de Stream Analytics (l’event hub) ainsi que la destination (notre base de données).

Nous pouvons désormais créer notre stream analytics.

Création du Job Azure Stream Analytics

Rendons nous sur le portail pour créer un nouveau job Stream Analytics :

Nous pouvons renseigner les éléments suivants :

Job Name : nom du job
Region : ceci correspond à la région d’exécution de votre processus de job. Actuellement il n’est disponible que les régions West Europe et Central US. Il faut tacher de placer son traitement au plus près de sa base de données et de son event hub
Storage Account : correspond aux comptes de stockages permettant de stocker les données de monitoring de vos Jobs. (dont notamment les informations sur les locks de polling sur les hubs, les dates d’exécution en cas de coupure etc…)

Une fois créer, nous avons besoin de définir les inputs/output et traitement :

Définition de l’input de job

Pour l’input, nous pouvons définir :

Un flux de données :
- a partir de blob
- a partir d’un event hub
Des données de références : pour effectuer des transcodifications (Id –> nom) lors du traitement par le job. Ces données de références sont des données de types CSV en général et enregistrer dans un blob

Nous choisissons ensuite notre type de données :

Et enfin nous configurons notre connexion à l’event hub :

Nous avons besoin de définir les éléments suivants :

input alias : sera utilisé comme nom de référence lors de l’écriture de notre requête pour le traitement du job
subscription : le choix de la souscription contenant l’event hub. Lorsque l’on choisit un event hub dans la souscription courant,l’ensemble des paramètres sont pré-remplis, sinon nous devons les remplir.
Service bus namespace : le namespace du service bus qui héberge l’event hub
Event hub Name : le nom du hub
Event Hub Policy Name : le nom de la règle d’accès pour générer les crédentials de connexion
Event Hub Policy Key : la clé de la règle d’accès pour générer les crédentials de connexion (NB: la règle doit avoir l’autorisation Manage pour fonctionner, cela devrait évoluer par la suite)
Event Hub Partition Count : le nombre de partition pour votre hub, ceci afin que le Job puisse créer autant de client que nécessaire pour récupérer l’ensemble des évènements.

Définition de l’output de job

Un seul output peut être déclaré. Les types d’output sont :

Blob Storage
Event Hub : peut permettre de chainer plusieurs JOB analytics ou de publier le résultat à plusieurs consommateur.
SQL Database

Pour le besoin de la démo, nous choisissons SQL Database :

Les paramètres a remplir sont les paramètres habituels de connexion à une base de données avec la table de destination des informations.

NB : vous pouvez rencontrer des problèmes de connexions à cet endroit, assurez vous que :

la table est bien crée
le CLUSTER INDEX REQUIS POUR SQL AZURE SOIT BIEN CREE !!!!! (important je me suis fait avoir sur ce point)

Définition de la requête

Enfin nous pouvons ensuite créer la requête d’agrégation (onglet query) :

 -- Job By Min
 SELECT DateAdd(second,-60,System.TimeStamp) as WinStartTime, system.TimeStamp as WinEndTime, 
 UserName,
 Scope,
 Sum(RequestLength) as TotalRequestLength,
 Sum(ResponseLength) as TotalResponseLength,  
 Count(*) as ApiCountCall 
    FROM event
    GROUP BY TumblingWindow(second, 60), Scope, UserName

Nous sommes dans une requete SQL standard ici, ou nous utilisons :

Sum et Count pour l’agrégation
- Nous récupérons la taille totale de données IN/OUT soumis par API/username
Group By pour le critère de regroupement

C’est la commande TumblingWindow(second, 60) qui va nous permettre d’agréger les données sur une Fenêtre de temps de 60 secondes (toutes les minutes).

Une fois configuré, et sauvegardé, nous pouvons démarrer notre job.

Dashboard

Nous pouvons monitorer le démarrage du Job à l’aide des journaux d’opérations de la plateforme Azure. Ils sont très utiles en cas d’erreur pour corriger. Un raccourci est disponible dans la partie Dashboard

La vue des logs :

Un exemple de log en erreur :

Résultat

Une fois démarré, nous pouvons appeler notre API et au bout d’une minute voir quelques renseignements dans la base de données que nous avons configuré.

Voila !

Dans le prochain billet, nous verrons la requête Stream Analytics à mettre en place pour pouvoir générer une alerte lorsque la consommation d’une API par un utilisateur dépasse un certains seuil.