Construire une stack data chez Wiidii

La donnée est primordiale pour le succès d’applications modernes, ce qui est d’autant plus vrai pour une startup qui doit être capable de s’adapter rapidement aux évolutions du marché.

Son utilisation est double : c’est un facteur déterminant pour le pilotage d’entreprise et de ses projets mais elle peut également être utilisée pour impacter directement l’expérience de nos utilisateurs via des outils de recommandation, de compréhension, etc.

La donnée étant une ressource indispensable, il est évident qu’il nous faut fournir les efforts nécessaires à son recueil, son stockage, sa manipulation et sa consommation au quotidien.

Dans cet article, nous allons décrire l’outillage mis en place chez Wiidii autour de la donnée, utilisant notamment les fonctionnalités de Databricks.

La stack technique

Le backend online de notre produit à évolué progressivement d’un monolithe vers une architecture microservice. Notre architecture est dite “DDD-oriented”, c’est-à-dire que chaque service va répondre à un problème indépendant dans un contexte clairement délimité (bounded context). Vous pouvez retrouver d’avantage d’information dans la méthodologie mise en place dans cet article de Microsoft sur le sujet. Les services communiquent entre eux indifféremment en REST ou grâce à un paradigme pub-sub via un bus de données partagé. Concernant la persistance de la donnée, les données d’un service ne sont accessibles que par les différentes instances de ce service (database per service pattern). L’objectif de la mise en place d’une stack data est donc notamment de pouvoir centraliser la donnée éclatée entre des dizaines de micro-services, implémentés via différents langages de programmation et avec plusieurs SGBD, pour pouvoir la manipuler simplement.

Architecture micro-services (Source : Wiidii)

En plus de cela, nous avons un certain nombre de contraintes à la mise en place de notre stack :

• Le recueil des données générées par un nouveau micro-service doit être le plus simple possible. Notre produit étant en constante évolution, nous ajoutons fréquemment de nouveaux domaines métier dont l’analyse est d’autant plus importante qu’ils sont généralement liés à de nouvelles fonctionnalités du produit dont on souhaite mesurer les impacts pour orienter les prochains développements dans le cadre d’une gestion de projet agile.
• La stack doit être capable de grossir en même temps que l’application et donc que l’architecture micro-services. Elle doit notamment pouvoir gérer plusieurs instances d’un même micro-service (scaling horizontal).
• La stack doit être robuste, auditable et respecter les contraintes réglementaires (RGPD notamment).

La stack historique

Dans les premières versions de notre produit, la problématique du recueil de données était déjà pris en compte. La solution qui avait été mise en place était d’avoir un micro-service spécifique qui servirait d’agrégateur de données puis de source de vérité unique pour la manipulation et l’analyse.

Pour ne pas impacter les services métier, ce service analytics s’abonnait à un maximum d’événements qui transitaient sur notre bus de données et s’en servait pour reconstruire la donnée véritable dans sa propre base.

Stack Data historique (Source : Wiidii)

Ce fonctionnement apporte des avantages certains, mais aussi de nombreux inconvénients.

Avantages :

• Donnée mise à jour en temps réel.
• Une seule infrastructure à gérer, à sécuriser, à faire scale (scaling vertical), etc.

Inconvénients :

• La maintenance est très chronophage. Le service analytics est en constante évolution en même temps que le reste des services : tout nouvel événement transitant sur le bus nécessite un abonnement supplémentaire et le traitement de celui-ci. Cela créé une adhérence énorme entre ce service et tous les autres et également un risque élevé d’oublis ou d’erreurs. Un nouvel événement qui n’est pas écouté par notre service analytics va créer au mieux un manque de données, au pire une incohérence par rapport à la vérité de l’application.
• Le service analytics ne sait qu’observer des échanges entre deux ou plusieurs services. Il ne sait pas explicitement ce que ces échanges impliquent pour les objets impactés. Par exemple si un service Message envoie sur le bus un événement suite à l’envoi d’un message par un utilisateur particulier, qui est susceptible de l’écouter ? Plusieurs services peuvent être abonnés à cet événement et modifier différents objets dans leurs bases respectives. Cela signifie que notre service centralisateur doit répliquer du code métier à appliquer à l’arrivée d’un événement, ce qui augmente encore l’adhérence entre les services, ainsi que le risque d’oubli.
• La donnée est difficilement manipulable en direct sans création et maintenance d’une gateway qui permettrait l’accès au data lake.

En pratique, ce fonctionnement demande une maintenance constante et très chronophage. La fiabilité de la donnée disponible est très fréquemment remise en question ce qui rend les analyses réalisées peu pertinentes. Il ne s’agit donc pas d’une solution viable à long terme.

Inverser la responsabilité

La première étape dans la mise en place de la nouvelle stack est l’inversion de la responsabilité dans l’identification des données à stocker dans le data lake.

Dans la version précédente, c’est le service qui faisait office de data lake qui avait cette responsabilité. C’est ce partage de responsabilités entre le service générant la donnée et le service l’écoutant qui posait un énorme problème de maintenance.

Dans la nouvelle version, c’est le service qui génère la donnée qui a la responsabilité de l’envoyer vers le data lake ou non. D’un point de vue de la séparation des responsabilités, cela a bien plus de sens : le data lake n’est là que pour stocker la donnée qu’on lui envoie et non pour choisir la donnée qu’il souhaite recevoir.

Au contraire, un service manipulant des objets spécifiques doit logiquement savoir dans quel contexte une donnée est dans un état susceptible d’être analysé.

Dans ce cadre, chaque service est susceptible d’envoyer un événement générique sur le bus — nommé AnalyticsEvent — qui contient le timestamp d’envoi, le service source, le type d’objet envoyé ainsi que le contenu de l’objet en question :

{
    "Timestamp": 1619356627,
    "Source": "User",
    "Payload": {
        "ObjectName": "User.Models.User",
	"Object": {
            "Id": "userId",
	    "FirstName": "John",
	    "LastName": "Snow",
	    "Email": "john.snow@foobar.com"
	}
}

Envoi unidirectionnel d’événements d’analytics (Source : Wiidii)

Le premier intérêt de ce fonctionnement est qu’il n’existe qu’un unique événement auquel le service consommateur doit s’abonner. Cela diminue grandement la maintenance à effectuer sur celui-ci.

Le second est que, étant donné qu’il s’agit d’un événement utilisé uniquement pour l’envoi vers notre data lake, il est possible de l’émettre via n’importe quel service sans impacter le fonctionnement de l’application.

Cela nous permet de rendre l’envoi de données — notamment par un nouveau service — extrêmement simple grâce à une surcharge de notre couche d’accès aux données (DAL). En effet, dans la majeure partie des cas, nous souhaitons analyser les données issues d’opérations en base de données : création, mise à jour, suppression. Il est très rare que nous souhaitions analyser un objet qui n’est pas persisté quelque part.

Ainsi, il suffit aux services d’ajouter un paramètre aux opérations en base pour envoyer sur le bus le résultat de cette opération :

public async User CreateUser(User user)
{
    return await userRepository.insert(user, publishEvent: true);
}

Il existe bien sûr un risque — lors du développement de nouvelles fonctionnalités notamment — d’oublier ce paramètre et donc d’avoir des données incomplètes dans l’environnement d’analyse. Cela fait partie des points d’attention lorsque nous effectuons des relecture de code. Selon le contexte, il est également envisageable d’avoir une DAL qui envoie par défaut de manière implicite le résultat de toutes les opérations sur le bus pour qu’elles soient stockées. Nous avons choisi de ne pas le faire, notamment pour réduire la quantité de donnée superflue envoyée et stockée.

Stocker la donnée

Nous avons donc un service unique qui reçoit — pratiquement en continu — des événements issus d’un grand nombre de services différents.

Indépendamment des actions que ce service effectue avec les données reçues, il est important de noter qu’il s’agit d’un service présent dans la même architecture que les autres. Il est donc assez simple de faire face à un pic d’événements reçus si besoin via un scaling horizontal ou vertical du service lui-même ou du bus d’événement utilisé selon l’endroit où se situe le goulot d’étranglement.

Nous avons choisi de donner à ce service une unique tâche pour les données reçues : les sortir de l’architecture vers un stockage cloud. En l’occurence, nous avons choisi d’utiliser un stockage Azure Data Lake Storage (ADLS) puisque nous souhaitions utiliser Azure Databricks et que cela nous permettait de rester dans un unique cloud.

Dans les faits, ce service — qui est finalement une sorte de gateway vers le cloud Azure — va simplement stocker chaque événement reçu sous la forme d’un fichier json dans un dossier spécifique de notre ADLS. Pour simplifier l’ingestion postérieure de la donnée, un dossier est créé par type d’objet stocké :

/ADLS
    /User.Models.User
        /1.json
        /2.json 
    /Chat.Models.Message
        /1.json
        /2.json

Ce stockage, qui n’est pas vraiment utilisable tel quel pour des analyses, nous sert de tampon et de stockage brut entre l’environnement applicatif et l’environnement d’analytics Databricks. Il est un event log exhaustif de toutes les opérations qui touchent aux objets.

Envoi de la donnée vers un stockage Cloud (Source : Wiidii)

Préparer les analyses

A ce point, nous avons donc une architecture micro-services qui est capable d’extraire en temps réel la donnée qu’elle créé ou transforme vers un stockage cloud.

L’étape suivante est d’ingérer le contenu de ce stockage dans un environnement qui contient les outils nécessaires pour réaliser des analyses.

Nous avons choisi Databricks, qui est une plateforme que nous utilisions déjà pour la fonctionnalité MLflow auparavant. Databricks propose, en plus de celle-ci, de nombreuses features indispensables pour centraliser la gestion de la donnée :

• Des bases de données managées et simples d’accès.
• Un environnement notebook qui permet de faire du prototypage.
• Une administration complète permettant une data governance efficace.
• Une fonctionnalité de planification et d’exécution de *jobs* réguliers ou manuels.

Pour ingérer le contenu de notre stockage cloud, nous utilisons la fonctionnalité AutoLoader de Databricks qui permet de faire du Spark Structured Streaming depuis un dossier spécifique d’un stockage cloud. AutoLoader met en place une Event Queue dans laquelle un événement est ajouté à chaque création de fichier dans le dossier ciblé. Cela nous permet d’ingérer des fichiers json brut dans une table Delta Lake sur Databricks. Le streaming présente deux intérêts majeurs ici :

• Tant que la consommation de la queue n’est pas lancée, les événements s’accumulent dans celle-ci. Cela permet beaucoup de flexibilité sur le temps entre deux ingestions. Il est tout à fait possible de faire du temps réel et de dépiler les événéments de queue dès qu’ils arrivent, tout comme il est possible de planifier le dépilage de la queue à intervalles réguliers. Il s’agit globalement d’un compromis à faire entre coûts d’infrastructure et délai d’ingestion.
• La queue ne contenant que les nouveaux événements, cela nous permet de ne pas avoir à recréer notre table de zéro à chaque nouvelle ingestion, mais uniquement d’ajouter les nouvelles lignes. C’est un gain de temps de calcul et donc d’argent qui n’est pas négligeable quand la quantité de données ingérée augmente.

Les premières versions que nous avions mises en place nécessitaient un schéma explicite pour lire de la donnée à partir de fichiers json, ce qui posait le problème de l’évolution de schéma pour éviter la perte de donnée lorsque l’objet concerné était modifié. La version 8.2 du Runtime Databricks apporte de nouvelles features sur cet aspect, et notamment la possibilité de faire de l’inférence de schéma et/ou de spécifier des type hints.

Nous utilisons les jobs Databricks pour planifier nos ingestions. Un job correspond à un type d’objet et donc à une table. Nous obtenons ainsi une table par objet dans laquelle chaque ligne correspond à un état d’un objet à un moment donné. Nous appelons cette donnée brute les tables bronze.

Exemple de table bronze (Source : Wiidii)

Ingestion de la donnée dans des tables bronze (Source : Wiidii)

Une fois la donnée brute disponible dans l’environnement Databricks, il ne reste plus qu’à la raffiner pour la rendre utilisable par des applications, qu’il s’agisse de ML, d’analyses ad-hoc ou de BI.

Ce raffinement se déroule en deux étapes : les tables silver et les tables gold.

Une table silver correspond au raffinement d’un objet dans un objectif spécifique. Elle ne prend en général en entrée qu’une seule table bronze. Plusieurs tables silver peuvent être générées à partir d’une même table bronze ; par exemple, à partir de la table User ci-dessus, on pourrait souhaiter générer :

• Une table qui correspond au dernier état connu de chaque objet. Il suffirait de prendre la ligne avec le Timestamp le plus élevé pour chaque Id.
• Une table qui calcule le temps entre chaque connexion d’un utilisateur. C’est dans ce type de cas que la conservation d’un event log complet prend tout son sens.

Il est à noter que Delta Lake permet d’utiliser une table comme source de streaming. Il est fréquent que nous utilisions cette fonctionnalité pour le peuplement des tables silver car cela permet de limiter les calculs redondants effectués.

Exemple de table silver (Source : Wiidii)

Ingestion dans des tables silver (Source : Wiidii)

Enfin, les tables gold correspondent à de la donnée correctement formatée et utilisable pour un usage downstream. Il s’agit en quelque sorte de l’équivalent des data mart d’un data warehouse. C’est à ce moment du processus que nous allons généralement effectuer des jointures entre plusieurs tables, pour croiser les informations issues de plusieurs objets. C’est également ici qu’a lieu une bonne partie du formatage de la donnée pour une meilleure exploitation selon le besoin métier sous-jacent. Nous privilégions généralement des tables gold avec des colonnes qui correspondent exactement à ce qui est attendu pour l’utilisation finale. Cela évite des tables massives et tentaculaires dont la documentation est très difficile et qui finissent par être abandonnées. On y préfère plus de tables avec une utilisation précise et correctement documentée.

Exemple de table gold (Source : Wiidii)

Pipeline de bout en bout jusqu’aux tables gold (Source : Wiidii)

Utiliser la donnée : PowerBI

Une fois les tables gold créées et leur actualisation planifiée, nous pouvons utiliser cette donnée pour impacter notre produit.

A l’heure actuelle, l’utilisation majeure que nous faisons de ces tables est la génération de reportings d’analyse via PowerBI.

PowerBI bénéficie d’un connecteur direct avec Databricks, ce qui simplifie fortement l’utilisation de nos données, notamment par de potentiels utilisateurs non-techniques.

Nous avons choisi de limiter autant que possible la quantité de calculs effectués dans PowerBI directement pour privilégier des calculs faits en amont qui profitent de la puissance de Spark ce qui nous permet de scale sur de gros volumes de données.

Intégration de PowerBI (Source : Wiidii)

Voici donc l’ensemble de la stack analytics telle que mise en place chez Wiidii. Celle-ci n’est évidemment pas parfaite, mais elle correspond à nos besoins en plus d’être relativement flexible.

Une amélioration que nous envisageons de mettre en place serait de s’éxonérer de stockage Cloud intermédiaire entre l’environnement applicatif et Databricks. Une solution pour ce faire serait de mettre en place une queue spécifique entre ces deux environnements. Cela nous permettrait de stocker les json envoyés tels quels directement dans une table Delta Lake qui serait un premier niveau de bronze. L’ensemble de la donnée, du plus brut au plus raffiné serait donc disponible dans Databricks directement alors qu’aujourd’hui nos tables bronze contiennent de la donnée qui a déjà passé une validation de schéma.

Merci d’avoir lu cet article jusqu’au bout ! Si vous avez des questions sur ce qui y a été présenté, n’hésitez pas à nous contacter !