Rafik BELLAHSENE, DATA ARCHITECT
Dans un précédent article, nous avons vu que Great Expectations est un outil pratique permettant de contrôler la qualité des données produites à la suite d’un processus ETL. Afin de superviser l’évolution quotidienne des données et appliquer des plans d’action automatisés en cas de non-conformité, il est crucial d’intégrer des rapports de validation de données.
Dans cet article, je vais vous expliquer comment optimiser vos processus ETL en intégrant des rapports de validation de données à l’aide de Great Expectations, Microsoft Azure et Office 365.
Au travers de ce billet, je vais vous exposer un exemple de process au sein des équipes data qui utilisent Great Expectations.
Le mécanisme simplifié est décrit comme suit :
- Nettoyage des données
- Enregistrement du rapport de qualité de données
- Envoi du rapport au Tech Lead de l’équipe (par e-mail)
- Création automatisée de tickets (Bugs) en cas de non-conformité de la donnée
Je pars du postulat que la phase de nettoyage des données a été réalisée en amont (vue dans l’article précédent). Je vais donc me focaliser sur les trois points restants.
Pour ce tutoriel, il est nécessaire d’avoir à disposition :
- Un espace de travail Azure Databricks
- Un Data Lake Storage Azure
- Une Azure Logic App
- Une organisation et un projet sur Azure DevOps
- Une messagerie Outlook (licence Office 365)
Enregistrement du rapport de qualité de données au sein du compte de stockage Azure
Reprenons là où nous en étions dans l’article précédent. Après avoir effectué l’opération de nettoyage de données, j’ai généré un rapport de qualité de données au format HTML, comme illustré dans la figure ci-dessous
# Validation des attentes
validation_result = ge_df.validate()
validation_result.meta["expectation_suite_name"] = expectation_suite.expectation_suite_name
# Formatage des résultats de validation au format HTML
validation_view = ValidationResultsPageRenderer().render(validation_result)
# Affichage des résultats
displayHTML(DefaultJinjaPageView().render(validation_view))
Par la suite, je procède à l’enregistrement du rapport dans le compte de stockage Azure afin de suivre l’évolution de la qualité des données au fil de nos différents processus ETL (cf. figure ci-dessous)
from datetime import datetime
from great_expectations.render.renderer import EmailRenderer,page_renderer
from great_expectations.checkpoint.util import send_webhook_notifications
import json
def save_file(**files):
"""
Fonction qui permet d'enregistrer les rapports au sein du compte de stockage dans un dossier YYYY/MM/DD/
"""
now = datetime.utcnow().strftime("%Y/%m/%d")
for filename,filecontent in files.items():
loc= f'/mnt/great_expectations/data_docs/{now}/{filename}'
try:
dbutils.fs.put(loc,filecontent)
except:
print("Overwriting Existing File")
dbutils.fs.rm(loc)
dbutils.fs.put(loc,filecontent)
# Enregistrement du rapport sous le nom de validation_results.html dans le dossier YYYY/MM/DD
save_file(**{
"validation_results.html": DefaultJinjaPageView().render(validation_view),
})
Le rapport de validation est stocké au sein du compte de stockage dans le dossier ‘data_docs/YYYY/MM/DD/validation_results.html‘
Envoi du rapport au Tech Lead de l’équipe
Restructuration des résultats de validation
Une fois que le rapport de validation de la qualité des données est créé et sauvegardé dans l’environnement Azure, nous sommes prêts à informer le Tech Lead de l’équipe Data et à lui présenter les résultats des tests d’intégrité qui ont été effectués.
Mais avant de continuer, il est nécessaire de restructurer les données de validation. Comme vous pouvez vous en douter, il est difficile d’envoyer un rapport HTML lisible par mail sans fournir le CSS.
Par conséquent, je vais reformater la structure de données via le code Python suivant, afin de la rendre plus exploitable :
import pandas as pd
new_df =pd.json_normalize(validation_result.to_json_dict()).explode("results")['results'].apply(pd.Series)
new_df = pd.concat([new_df.drop(columns = ["result",'expectation_config']),
new_df['result'].apply(pd.Series),
new_df['expectation_config'].apply(pd.Series)], axis=1)
new_df['level'] = new_df['observed_value'].fillna("Column")
new_df['level'] = ["Table" if x != 'Column' else "Column" for x in new_df['level']]
new_df = new_df[['level','expectation_type', 'kwargs', 'success','element_count',
'missing_count','missing_percent','unexpected_count',
'unexpected_percent','unexpected_percent_total',
'observed_value']].\
rename(columns={"expectation_type":"type",
"kwargs": "details",
"success" : "succeeded",
"missing_count":"null_count",
"missing_percent" : "null_percent",
"unexpected_count" : "unexpected_non_null_values",
"unexpected_percent":"unexpected_total_percent",
"unexpected_percent_total":"unexpected_non_null_percent",
"expectation_level" : "level"
})
Dans ce code, j’utilise la librairie pandas pour transformer les données de validation (enregistrées au format HTML initialement) en un dataframe structuré et facilement exploitable.
Je crée ensuite un nouveau dataframe « new_df » à partir du JSON produit par validation_result.to_json_dict.
Je réorganise le dataframe et renomme certaines colonnes pour faciliter la compréhension des données.
Le rapport finalisé, passons à la configuration de l’envoi automatique d’e-mail et création d’issues dans DevOps.
Configuration de la Logic Apps
Définition de Azure Logic Apps
Azure Logic Apps est un service serverless permettant de créer et de concevoir des workflows automatisés. Il est utilisé dans la majorité des organisations et des scénarios B2B pour développer des solutions d’intégration à grande échelle (Cf. Documentation).
Explication du processus fonctionnel
Le process de notification s’articule comme suit :
- Suite à la génération des rapports de qualité de données, nous vérifions s’ils comportent des attentes non validées. Si tel est le cas, nous informons le Tech Lead par e-mail. Cette notification s’effectue à l’aide d’un webhook que nous appelons via une requête HTTPS en méthode POST depuis notre notebook Databricks. L’ensemble des informations transmises par la requête HTTPS est inclus dans l’e-mail.
- Le responsable détermine s’il est nécessaire de signaler une non-conformité aux équipes de développement. Si tel est le cas, un ticket sera automatiquement généré au sein du projet Azure DevOps de son équipe.
Développement du processus
La Logic Apps est un composant serverless disposant de 2 modes de déclenchement :
- Déclenchement sur ordonnancement : C’est ce qu’on appelle des « Time Trigger»
- Déclenchement sur évènement : C’est ce qu’on appelle des « Event Trigger»
Dans notre cas, nous allons utiliser un déclencheur sur événement de type appel http, ce qui permet d’exécuter le workflow défini dans la Logic Apps en faisant un appel http depuis notre notebook databricks.
Nous allons donc commencer par la configuration du déclencheur HTTP :
{
"kind": "Http",
"inputs": {
"schema": {
"properties": {
"details": {
"type": "string"
},
"message": {
"type": "string"
},
"object": {
"type": "string"
}
},
"type": "object"
}
}}
Dans la figure ci-dessus, nous définissons le schéma des entrées (inputs).
Celui-ci se compose des 3 paramètres suivants :
- Object : Correspond à l’objet du mail que l’on souhaite envoyer au Tech-Lead
- Message : Correspond au corps du mail
- Details : Correspond au contenu du bug que l’on souhaite afficher sur Azure DevOps
Ils sont ensuite utilisés dans l’action « Envoyer un e-mail avec des options » (cf. Figure ci-dessous).
Pour configurer cette action, il est nécessaire d’indiquer :
- Le destinataire (To)
- Les options utilisateurs (User Options)
- L’objet du mail (Subject)
- Le contenu du mail (Body)
Vous noterez que j’ai employé les paramètres « Object » et « Message » définis précédemment, et créé deux options utilisateur (Créer un bug et Refuser l’action). Ces dernières seront affichées dans le corps de l’e-mail sous forme de boutons cliquables, et vont nous permettre de décider de créer un ticket ou non dans l’organisation Azure Devops. Il est donc important de vérifier la valeur retournée par le clic de l’utilisateur (Figure ci-dessous).
Dans le cas où l’utilisateur clique sur « Create bug« , nous générerons un « Work Item » dont le contenu est récupéré depuis le paramètre « details » de la requête HTTPS. Le ticket est créé au sein d’un projet Azure DevOps.
Vous remarquerez que j’ai utilisé les paramètres « object » et « details » du déclencheur HTTP pour définir le titre et le contenu du ticket Azure Devops.
Envoi de la requête depuis le notebook Databricks
A présent, notre worfklow Logic Apps est prêt. Nous pouvons donc l’appeler directement via notre notebook Databricks. Pour ce faire, nous allons utiliser la fonction « send_webhook_notifications » de Great Expectations.
Cette fonction prend en paramètre :
- L’URL du webhook : Qui correspond à l’URL du déclencheur http de la Logic APPS
- Le message ou la donnée à envoyer via la requête POST : Qui correspond aux paramètres qu’on avait défini dans le déclencheur http.
logic_app_url = “YOUR LOGIC APPS URL”
for res in validation_result.results :
if (not res.success): # Si un résultat de validation a un statut Success == False
send_webhook_notifications(webhook = logic_app_url,
query = {
# Objet du mail
"object" : EmailRenderer().render(validation_result)[0],
# Contenu du mail
"message" : EmailRenderer().render(validation_result)[1],
# Contenu du ticket (Bug)
"details" : new_df.to_markdown(index=False)
},
target_platform ="logic_app")
Pour trouver l’URL de la logic apps, il suffit de cliquer sur le déclencheur HTTP et copier la valeur (Cf. Figure ci-dessous) :
Une fois la requête envoyée, le responsable recevra le mail suivant pour le notifier de la non-conformité des données et décidera ainsi de créer un ticket ou non.
S’il clique sur « Create Bug », le ticket sera automatiquement créé au sein de l’organisation Azure Devops (cf. Figure ci-dessous)
Au travers de ce diptyque, nous avons examiné les multiples cas d’utilisation de Great Expectations, démontrant ainsi sa grande polyvalence et sa facilité d’intégration dans des environnements de production variés. Il demeure une solution pratique pour gérer et valider des données à grande échelle, en permettant de détecter rapidement les anomalies et les erreurs au sein de processus ETL. De plus, sa capacité à s’intégrer facilement avec d’autres technologies existantes le rend très adaptable et personnalisable, ce qui lui permet de répondre aux besoins spécifiques de chaque entreprise.
En conclusion, Great Expectations est un outil de qualité qui offre une solution pratique pour la gestion de données, avec un potentiel d’application étendu dans de nombreux domaines d’activité.