Les bonnes pratiques de tests unitaires PySpark

Dans ce nouvel article autour du Craft, nous vous proposons de vous intéresser aux tests unitaires PySpark.

Rappelons d’abord de quoi il s’agit :  les tests unitaires en PySpark sont une façon de tester le code en vérifiant le comportement de chaque partie individuellement. Pytest est une bibliothèque de tests utilisée pour faciliter la création et l’exécution de tests unitaires en PySpark. Ces derniers sont importants, car ils permettent de détecter rapidement les erreurs de code, et de s’assurer que les résultats produits par le code sont corrects et cohérents. Aussi, les tests unitaires aident à garantir la qualité et la fiabilité du code.

Faut-il être un testeur pro ?

Non, rassurez-vous. Cet article ne s’adresse pas aux testeurs professionnels, mais à tous les profils Data qui cherchent à améliorer la qualité de leurs codes sur Databricks/PySpark. Le but de cet article est de montrer comment intégrer les tests unitaires sur des projets Data en fournissant des outils prêts à consommer plutôt que de coder des fonctionnalités complexes et réinventer la roue. Nous le verrons plus en détail dans les parties suivantes, mais gardez en tête qu’il existe déjà plusieurs outils permettant de faire des tests unitaires.

Quels sont les prérequis ?

Pour débuter, il faudra avoir un environnement de test isolé afin de garantir que la phase des tests se déroule en amont des environnements de production.

Donc nous considérons que les pipeline CI/CD de l’environnement de test sont déjà mis en place. Nous aurons ensuite besoin de ces outils :

• Un IDE Python et Spark, tel que VS Code.
• La librairie de tests Pytest, qui fournit un framework de tests en Python.

Tests unitaires et débogage des notebooks Databricks : comment ne plus confondre les deux ?

D’une façon simple, le débogage consiste à trouver et corriger les erreurs dans le code, alors que les tests unitaires sont des petits bouts de code qui vérifient si chaque partie du code fonctionne correctement.

Le débogage consiste donc à chercher les erreurs, et c’est ce qu’on a l’habitude de faire à travers des notebooks Databricks, alors que les tests unitaires sont là pour vérifier si tout fonctionne comme prévu.

Dans notre cas, on aimerait tester les fonctions codées dans les notebooks, et vérifier qu’elles fonctionnent comme prévu avant déploiement. Cela nous permettra de savoir s’il y a eu une régression pendant les développements où il n’y a pas eu d’impacts sur l’existant.

Et pour cela, on va convertir ce qui est développé et validé dans les notebooks en fonctions dans des scripts Python.

Finalement, les tests unitaires se feront sur les fonctions converties en script qu’on vient de créer.

Par où commencer ?

La toute première partie consistera à initier notre projet Python, et à le découper en dossiers, voire en sous-dossiers pour qu’on puisse se repérer assez facilement pendant les étapes suivantes.

• .cicd: c’est le folder qui contient la définition des pipelines CI/CD.
• Pipeline:
  • yml : ce fichier contiendra comment on va lancer nos tests unitaires dans la CI, comme nous le verrons dans la toute dernière partie de cet article.
  • yml : il s’agit de templates de déploiement, mais qui ne rentrent pas dans notre contexte vu que les tests unitaires seront intégrés sur la CI.
• Templates
  • yml : dans ce fichier, on définira dans la CI le lancement des tests unitaires et le coverage, puis la publication de leurs résultats.
  • Template_component : au même niveau que le .cicd, c’est dans ce dossier qu’on définira les fonctions à inclure dans les tests unitaires, ainsi les données à utiliser pendant les tests et leurs résultats.

La première chose qu’on remarque dans le folder « Template » c’est qu’il y a plusieurs fichiers __init__.py. Ce sont des fichiers de configuration exigés par Python pour prendre en considération les dossiers lors de la génération des wheels (une wheel est le livrable de Python, comme les jars en Java ou les DLL en .NET).

Ce folder contiendra les fonctions de tests et de transformations que nous coderons dans les parties suivantes dans cet article. Ce folder est basé par zone de traitement à laquelle nous faisons référence : bronze, silver, gold. Par exemple dans le folder silver, on trouvera les fonctions et tests unitaires qu’on établit et qu’on a intégrés à la zone de traitement silver.

Quoi faire après ?

Avant d’écrire nos tests unitaires, nous allons créer une session Spark que nous pourrons réutiliser dans tous nos tests. Nous créons donc des fixtures PyTest dans le fichier conftest.py.

Qu’est-ce qu’une fixture ?

Les fixtures sont des fonctions de la librairie Pytest qui permettent de gérer les états et les dépendances de nos applications. Elles sont utiles pour fournir des données de test et une variété de types de valeurs lorsqu’elles sont explicitement appelées par notre logiciel de test. On peut utiliser les données simulées que les fixtures créent pour plusieurs tests. C’est très utile pour des objets complexes comme la session Spark qui met un temps de création significatif.

Le code suivant est un exemple de la façon de configurer une session SparkSession réutilisable à l’aide de la bibliothèque Pytest.

Cette session est configurée pour stocker temporairement toutes les tables Delta dans un répertoire temporaire. Pour cela, nous utilisons des fixtures Pytest, comme expliqué précédemment, qui créent des objets une fois et les réutilisent dans de multiples tests.

Notre premier exemple est une fixture qui crée la SparkSession pour tous les tests unitaires et une deuxième fixture qui permet de récupérer le répertoire temporaire utilisé pour stocker toutes les tables Delta :

Le dernier exemple est une autre fixture qui recherche un dossier avec le même nom que le module de test, et, s’il est disponible, déplace tous les contenus vers un répertoire temporaire pour que les tests puissent les utiliser librement :

Mais une fois les fixtures définies, comment les utiliser ?

C’est le vif du sujet 😊, l’étape suivante, c’est l’écriture des fonctions des tests unitaires.

Dans les paramètres des fonctions, on fera appel aux fixtures déjà instanciées précédemment tels que la Spark session. Dans ce qui suit, vous trouverez des exemples de fonctions qu’on aimerait tester à chaque fois qu’on déroule un scénario de test.

Un premier test, test_return_first_column_of_df() est une fonction qui prend en entrée un DataFrame Spark avec plusieurs colonnes et renvoie un DataFrame Spark ne contenant que la première colonne. Le test compare le DataFrame renvoyé par la fonction à un DataFrame au format attendu :

Un deuxième exemple d’une fonction qui prend une chaîne de caractères au format YYYYMMDDhhmmss et la convertit en objet datetime. Le test compare la valeur renvoyée par la fonction à une valeur attendue.

Une exception est levée lorsque la chaîne de caractères fournie n’est pas au format attendu :

Le troisième test est test_write_delta_table(). On teste une fonction qui écrit un DataFrame Spark dans une table Delta, puis lit la table et compare le résultat avec un DataFrame attendu. Le chemin de la table Delta et la SparkSession sont fournis en paramètres via les fixtures qu’on a définies dans la partie précédente.

Mais quelles données utiliser pour réaliser ces tests et à quel format ?

C’est simple : pour les scénarii simples tels que l’exemple des dates ci-dessus, nous avons défini les données dans les tests eux-mêmes.

Pour les tests se basant sur les dataframes, la bonne pratique consiste à utiliser des fichiers d’extension simples tels que CSV ou Json qui seront plus faciles à lire et à manipuler par un utilisateur simple. Il ne faut pas oublier qu’on doit minimiser la complexité et le temps d’exécution des tests unitaires, car leur objectif reste de s’assurer que le code continue de fonctionner correctement au fur et à mesure que nous apportons des modifications.

Et après toutes ces configurations, comment lancer ces tests unitaires ?

On pourra simplement les lancer directement sur l’IDE en local, et observer les résultats de tests unitaires et coverage. Dans l’exemple de la capture de test suivante, il y a eu 4 tests unitaires : deux en succès et deux ont fini en échecs. En cliquant sur les tests en échecs, on pourra avoir les logs d’échecs des tests qui ont échoué :

L’objectif de ces tests unitaires reste de les automatiser et les intégrer dans l’approche CI/CD. Dans cette partie, nous ferons les configurations pour ajouter un stage de test dans la CI.

On va initier un fichier test_component.yml qui sera le fichier de configuration pour le pipeline de tests unitaires du projet.

On commence par déclarer une liste de paramètres qui seront utilisés dans le pipeline :

parameters:
- name: component_directory
  type: string
- name: build_name
  type: string
- name: artifactory
  type: object
- name: depends_on
  type: string
  default: '[]'

Puis on définit un premier job qui définit la version de Python à utiliser, le chemin d’où on récupère le code source, puis on installe les dépendances spécifiées dans “requirements.txt” :

Le job exécute ensuite des tests unitaires à l’aide de la bibliothèque Pytest et publie les résultats de tests et de coverage :

Dans requirement.txt, on définit les librairies dont nous aurons besoin dans notre projet et nos tests :

Une fois intégrés dans le CI, les tests unitaires se lanceront implicitement au moment du lancement du pipeline CI.

On pourra tout de même lancer les tests sur l’IDE en local, et observer les résultats de tests unitaires et coverage.

On voit dans le résultat d’exécution qu’il y a eu sept tests : le premier a fini en succès, dans le deuxième il manque les lignes 4 à 64 qui n’ont pas été testées, etc.

En savoir plus sur le Craft

Vous souhaitez en savoir plus sur le Craftsmanship ? Retrouvez notre série d’articles publiés à l’occasion du Mois du Craft :

• Le Craft est-il toujours d’actualité ?
• Comment choisir l’architecture logicielle idéale grâce aux Architectural Drivers ?
• Comment crafter une infrastructure avec Terraform ?
• Craft et PowerShell : de la nécessité d’appliquer les pratiques de Software Engineering à l’infrastructure
• Télémétrie : la garantie d’un code qui fonctionne
• Boostez la performance de vos applications avec Asyncio : un guide pratique pour les développeurs Python