Azure Stream Analytics est un moteur de traitement des évènements en temps réel. Grâce à lui vous pouvez gérer simplement plusieurs flux d'entrée (comme un Event Hub) et de sortie (Storage Azure, base de donnée, etc.) avec la possibilité de les rediriger, monitorer, ou transformer instantanément.

Présentation

Le service se compose de 3 parties  :

• Input

Il s’agit du ou des services Azure d’où le Stream Analytics récupèrera des données .

Les types d’entrées possibles sont :

• Event Hub
• Blob Storage
• IoT Hub

• Output

Il s’agit des services Azure dans lesquels notre Stream Analytics va écrire les données transformées.

Les types de sorties possibles sont :

• SQL Database
• Blob Storage
• Event Hub
• Table storage
• Service bus queue
• Service bus topic
• DocumentDb
• Power BI

• Query

Il s’agit du cerveau de la ressource. Utilisant une syntaxe SQL-like de requêtage de flux, ce langage permet de très facilement manipuler les flux entrant de Stream Analytics pour les rediriger vers les flux sortant. Les possibilités d'utilisation sont très larges : redirection, agrégation, transformation des données ou même création d'alertes.

Utilisation

A titre d’exemple, nous allons créer un Stream Analytics avec en entrée un IoT Hub, et trois sorties : 2 Blob Storage et un Event Hub.

L'objectif sera de récupérer les données de capteur de température envoyées dans un IoT Hub et de les rediriger dans un Event Hub pour une consommation instantanée ainsi que d'archiver toutes nos données dans un Blob Azure sous 2 formats différents.

• Provisioning

Pour commencer, nous allons créer (déployer) notre Stream Analytics.

Pour ce faire, depuis le nouveau portail Azure, on clique sur “ajouter” et on recherche "Stream analytics", puis on sélectionne "Stream Analytics Job". On choisit enfin un nom et on crée notre ressource.

Nous allons maintenant paramétrer nos entrées et sorties.

• Configuration de l’input

Pour ce faire on choisit input, puis on sélectionne le type IoT Hub,

Plusieurs choses sont à configurer :

• Input alias : très important car il servira à designer cette entrée dans nos queries
• Source Type : 2 choix possibles. Data Stream correspond à des flux continus alors que Référentiel représente des flux statiques (Storage Blob)
• Source : comme expliqué plus haut plusieurs choix sont possibles. Ici nous utiliserons le IoT Hub
• Event serialization format : Avro, Csv ou JSON

Avro est un système de sérialisation de données permettant un échange compact de données sous format binaire, il est notamment utilisé pour Apache Hadoop.

Pour se connecter au hub il nous faut les informations suivantes :

• IoT hub,
• Shared access policy name,
• Shared access policy key,
• Consumer group.

Pour savoir où récupérer ces infos, je vous conseille de lire l'article suivant (Ici).

• Configuration des Outputs

Maintenant nous allons configurer plusieurs outputs. Tout d'abord l'Event Hub : un autre article pour en savoir plus sur le sujet (Ici)

Configurons ensuite 2 sorties vers un Azure Blob Storage, la première pour archiver nos données en JSON et la seconde pour l’enregistrement en CSV.

On retrouve nos habituels “Storage Account", "Storage account key" et le container.

La sauvegarde en JSON pourra être effectuée selon un path dynamique qui créera un blob par jour.

La sortie au format CSV sera stockée dans un Blob par jour que nous pourrons éventuellement utiliser comme données de test avec le service de machine Learning Microsoft : Azure ML.

Tout ce beau monde pouvant à présent communiquer, il est temps d'attaquer les choses sérieuses avec la partie logique.

• Query

Les queries vont nous permettre de manipuler directement les flux de nos entrées/sorties de notre Stream Analytics.

Par exemple, ici IoT Hub nous envoie des données provenant de plusieurs capteurs sous la forme JSON suivante :

[{"data":{"deviceId":"myFirstDevice","temperature":29.22772163582394},"EventProcessedUtcTime":"2016-02-07T16:00:03.4759784Z","PartitionId":3,"EventEnqueuedUtcTime":"2016-02-07T16:00:02.6180000Z","IoTHub":{"MessageId":null,"CorrelationId":null,"ConnectionDeviceId":"myFirstDevice","ConnectionDeviceGenerationId":"635899211883606148","EnqueuedTime":"0001-01-01T00:00:00.0000000","StreamId":null}}] 

Pour commencer, nous voulons simplement rediriger les données dans un Event Hub.

Voici donc la requête associée :

SELECT 
    * 
INTO 
    [EventHub]
FROM 
    [iotHubInput] 

La query parle d'elle-même. On récupère les informations voulues de notre objet JSON depuis l'IoT hub et on l'envoi tel quel dans notre Event Hub.

Le Stream Analytics n’a en réalité aucune connaissance du contenu du message et n’est dans ce cas qu’un messager. Le langage permet aussi la création d'alertes.

Par exemple je veux que si la température d'un capteur dépasse 100° un objet d'alerte soit envoyé à l'Event Hub.

  Nous rajoutons alors une condition SQL sous cette forme :

SELECT 
    'Alert', 
    data.Temperature, 
    data.deviceId 
WHERE data.Temperature > 100 
INTO [EventHub] 
FROM [iotHubInput] 

Nous souhaitons aussi archiver toute les données de nos capteurs de température. Nous faisons la même chose que pour l'Event Hub:

SELECT 
    *
INTO 
    [TempArchive]
FROM 
    [iotHubInput] 

Ici on changera seulement le INTO pour diriger le flux vers notre output Blob Storage.

Il est aussi possible de rediriger seulement une partie des informations dans un CSV. Pour cela nous devons connaitre la structure du message entrant pour en extraire les propriétés voulues :   

SELECT 
    data.deviceId AS deviceId,
    data.Timestamp AS TimeStamp, 
    data.Temperature AS Temperature, 
    data.Longitude AS Longitude, 
    data.Latitude AS Latitude
INTO 
    [AzureML]
FROM 
    [iotHubInput]

Tout ceci est très simple d'utilisation. Une fois notre job lancé, Azure Stream Analytics s'occupe de tout.

On peut effectivement constater que nos fichiers ont bien été créé dans le Blob.

Conclusion

Le service Azure Stream Analytics est donc un outil très puissant pour gérer une énorme quantité de flux continus. Il est particulièrement efficace dans un contexte IoT pour rediriger, manipuler et transformer tous les flux provenant de millions de capteurs simultanés tout en restant très simple d'utilisation.