Aller au contenu principal
Dernière mise à jour le

Tests unitaires pour les notebooks Databricks

Vous pouvez utiliser les *tests unitaires* pour améliorer la qualité et la cohérence du code de vos notebooks. Les tests unitaires sont une approche pour tester des unités de code autonomes, telles que des fonctions, dès le début et fréquemment. Cela vous aide à trouver plus rapidement les problèmes dans votre code, à découvrir plus tôt les hypothèses erronées concernant votre code et à rationaliser vos efforts de codage globaux.

Cet article est une introduction aux tests unitaires de base avec des fonctions. Les concepts avancés tels que les classes et interfaces de test unitaire, ainsi que l'utilisation de stubs, de mocks et de harnais de test, bien que également pris en charge lors des tests unitaires pour les notebooks, dépassent le cadre de cette page. Cet article ne couvre pas non plus d'autres types de méthodes de test, comme les tests d'intégration, les tests système, les tests d'acceptation ou les tests non fonctionnels, tels que les tests de performance ou les tests d'ergonomie.

Cet article démontre ce qui suit :

  • Comment organiser les fonctions et leurs tests unitaires.
  • Comment écrire des fonctions en Python, R, Scala, ainsi que des fonctions définies par l’utilisateur en SQL, qui sont bien conçues pour être testées unitairement.
  • Comment appeler ces fonctions à partir de Notebooks Python, R, Scala et SQL.
  • Comment écrire des tests unitaires en Python, R et Scala en utilisant les frameworks de test populaires pytest pour Python, testthat pour R et ScalaTest pour Scala. Également comment écrire du SQL qui teste unitairement les fonctions SQL définies par l'utilisateur (SQL UDFs).
  • Comment exécuter ces tests unitaires à partir de Notebooks Python, R, Scala et SQL.
remarque

Databricks recommande d'écrire et d'exécuter vos tests unitaires dans un Notebook. Alors que vous pouvez exécuter certaines commandes dans le terminal web, le terminal web présente plus de limitations, telles qu'un manque de prise en charge pour Spark. Consultez Exécuter des commandes Shell dans le terminal web Databricks.

Organiser les fonctions et les tests unitaires​

Il existe plusieurs approches courantes pour organiser vos fonctions et leurs tests unitaires avec des Notebooks. Chaque approche présente ses avantages et ses défis.

Pour les notebooks Python, R et Scala, les approches courantes incluent les suivantes :

  • Stockez les fonctions et leurs tests unitaires en dehors des Notebooks..

    • Avantages : Vous pouvez appeler ces fonctions avec des Notebooks et en dehors. Les frameworks de test sont mieux conçus pour exécuter des tests en dehors des Notebooks. Databricks fournit une suite d'outils pour découvrir, exécuter et suivre les tests unitaires Python directement dans le workspace. Consultez les tests unitaires Python dans le Workspace.
    • Défis : Cette approche n'est pas prise en charge pour les notebooks Scala. Cette approche augmente également le nombre de fichiers à suivre et à maintenir.

  • Stockez les fonctions dans un Notebook et leurs tests unitaires dans un Notebook séparé..

    • Avantages : ces fonctions sont plus faciles à réutiliser dans tous les Notebooks.
    • Défis : Le nombre de notebooks à suivre et à maintenir augmente. Ces fonctions ne peuvent pas être utilisées en dehors des notebooks. Ces fonctions peuvent également être plus difficiles à tester en dehors des Notebook.

  • Stockez les fonctions et leurs tests unitaires dans le même notebook..

    • Avantages : les fonctions et leurs tests unitaires sont stockés dans un seul Notebook pour faciliter le suivi et la maintenance.
    • Défis : ces fonctions peuvent être plus difficiles à réutiliser entre les Notebooks. Ces fonctions ne peuvent pas être utilisées en dehors des notebooks. Ces fonctions peuvent également être plus difficiles à tester en dehors des Notebook.

Pour les notebooks Python et R, Databricks recommande de stocker les fonctions et leurs tests unitaires en dehors des notebooks. Pour les notebooks Scala, Databricks recommande d'inclure les fonctions dans un notebook et leurs tests unitaires dans un notebook distinct.

Pour les Notebooks SQL, Databricks vous recommande de stocker les fonctions en tant que fonctions définies par l'utilisateur SQL (UDF SQL) dans vos schémas (également appelés bases de données). Vous pouvez ensuite appeler ces UDF SQL et leurs tests unitaires à partir de Notebooks SQL.

Fonctions d'écriture​

Cette section décrit un ensemble simple d'exemples de fonctions qui déterminent les éléments suivants :

  • Indique si une table existe dans une base de données.
  • Indique si une colonne existe dans une table.
  • Combien de lignes existent dans une colonne pour une valeur à l'intérieur de cette colonne.

Ces fonctions sont conçues pour être simples, afin que vous puissiez vous concentrer sur les détails des tests unitaires de cette page plutôt que sur les fonctions elles-mêmes.

Pour obtenir les meilleurs résultats de tests unitaires, une fonction doit renvoyer un résultat unique et prévisible et être d'un seul type de données. Par exemple, pour vérifier si quelque chose existe, la fonction doit renvoyer une valeur booléenne de vrai ou faux. Pour renvoyer le nombre de lignes existantes, la fonction doit renvoyer un nombre entier non négatif. Il ne doit pas, dans le premier exemple, renvoyer faux si quelque chose n'existe pas, ni la chose elle-même si elle existe. De même, pour le second exemple, il ne doit pas renvoyer le nombre de lignes existantes, ni faux si aucune ligne n'existe.

Vous pouvez ajouter ces fonctions à un Workspace Databricks existant comme suit, en Python, R, Scala ou SQL.

Le code suivant suppose que vous avez configuré l'intégration Git pour les dossiers Git, ajouté un référentiel, et que le référentiel est ouvert dans votre workspace Databricks.

Créez un fichier nommé myfunctions.py dans le dépôt, et ajoutez le contenu suivant au fichier. D'autres exemples sur cette page s'attendent à ce que ce fichier soit nommé myfunctions.py. Vous pouvez utiliser différents noms pour vos propres fichiers.

Python
import pyspark
from pyspark.sql import SparkSession
from pyspark.sql.functions import col

# Because this file is not a Databricks notebook, you
# must create a Spark session. Databricks notebooks
# create a Spark session for you by default.
spark = SparkSession.builder \
                    .appName('integrity-tests') \
                    .getOrCreate()

# Does the specified table exist in the specified database?
def tableExists(tableName, dbName):
  return spark.catalog.tableExists(f"{dbName}.{tableName}")

# Does the specified column exist in the given DataFrame?
def columnExists(dataFrame, columnName):
  if columnName in dataFrame.columns:
    return True
  else:
    return False

# How many rows are there for the specified value in the specified column
# in the given DataFrame?
def numRowsInColumnForValue(dataFrame, columnName, columnValue):
  df = dataFrame.filter(col(columnName) == columnValue)

  return df.count()

Appeler des fonctions​

Cette section décrit le code qui appelle les fonctions précédentes. Vous pourriez utiliser ces fonctions, par exemple, pour compter le nombre de lignes dans une table où une valeur spécifiée existe dans une colonne spécifiée. Cependant, vous voudriez vérifier si la table existe réellement, et si la colonne existe réellement dans cette table, avant de continuer. Le code suivant vérifie ces conditions.

Si vous avez ajouté les fonctions de la section précédente à votre workspace Databricks, vous pouvez appeler ces fonctions depuis votre workspace comme suit.

Créez un Notebook Python dans le même dossier que le fichier myfunctions.py précédent de votre référentiel, et ajoutez le contenu suivant au Notebook. Modifiez les valeurs des variables pour le nom de la table, le nom du schéma (base de données), le nom de la colonne et la valeur de la colonne selon les besoins. Ensuite, associez le Notebook à un cluster et exécutez le Notebook pour voir les résultats.

Python
from myfunctions import *

tableName   = "diamonds"
dbName      = "default"
columnName  = "clarity"
columnValue = "VVS2"

# If the table exists in the specified database...
if tableExists(tableName, dbName):

  df = spark.sql(f"SELECT * FROM {dbName}.{tableName}")

  # And the specified column exists in that table...
  if columnExists(df, columnName):
    # Then report the number of rows for the specified value in that column.
    numRows = numRowsInColumnForValue(df, columnName, columnValue)

    print(f"There are {numRows} rows in '{tableName}' where '{columnName}' equals '{columnValue}'.")
  else:
    print(f"Column '{columnName}' does not exist in table '{tableName}' in schema (database) '{dbName}'.")
else:
  print(f"Table '{tableName}' does not exist in schema (database) '{dbName}'.")

Écrire des tests unitaires​

Cette section décrit le code qui teste chacune des fonctions décrites au début de cette page. Si vous apportez des modifications aux fonctions à l’avenir, vous pouvez utiliser les tests unitaires pour déterminer si ces fonctions fonctionnent toujours comme vous vous y attendez.

Si vous avez ajouté les fonctions vers le début de cette page à votre workspace Databricks, vous pouvez y ajouter des tests unitaires comme suit.

Créez un autre fichier nommé test_myfunctions.py dans le même dossier que le fichier myfunctions.py précédent de votre repo, et ajoutez le contenu suivant au fichier. Par défaut, pytest recherche .py fichiers dont les noms commencent par test_ (ou se terminent par _test) pour les tester. De même, par défaut, pytest examine ces fichiers à la recherche de fonctions dont les noms start par test_ pour les tester.

En général, il est préférable de *ne pas* exécuter de tests unitaires sur des fonctions qui traitent des données en production. Ceci est particulièrement important pour les fonctions qui ajoutent, suppriment ou modifient de toute autre manière les données. Pour protéger vos données de production contre toute altération inattendue par vos tests unitaires, vous devez exécuter des tests unitaires sur des données non-production. Une approche courante consiste à créer des données fictives aussi proches que possible des données de production. L'exemple de code suivant crée des données fictives pour l'exécution des tests unitaires.

Python
import pytest
import pyspark
from myfunctions import *
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, IntegerType, FloatType, StringType

tableName    = "diamonds"
dbName       = "default"
columnName   = "clarity"
columnValue  = "SI2"

# Because this file is not a Databricks notebook, you
# must create a Spark session. Databricks notebooks
# create a Spark session for you by default.
spark = SparkSession.builder \
                    .appName('integrity-tests') \
                    .getOrCreate()

# Create fake data for the unit tests to run against.
# In general, it is a best practice to not run unit tests
# against functions that work with data in production.
schema = StructType([ \
  StructField("_c0",     IntegerType(), True), \
  StructField("carat",   FloatType(),   True), \
  StructField("cut",     StringType(),  True), \
  StructField("color",   StringType(),  True), \
  StructField("clarity", StringType(),  True), \
  StructField("depth",   FloatType(),   True), \
  StructField("table",   IntegerType(), True), \
  StructField("price",   IntegerType(), True), \
  StructField("x",       FloatType(),   True), \
  StructField("y",       FloatType(),   True), \
  StructField("z",       FloatType(),   True), \
])

data = [ (1, 0.23, "Ideal",   "E", "SI2", 61.5, 55, 326, 3.95, 3.98, 2.43 ), \
         (2, 0.21, "Premium", "E", "SI1", 59.8, 61, 326, 3.89, 3.84, 2.31 ) ]

df = spark.createDataFrame(data, schema)

# Does the table exist?
def test_tableExists():
  assert tableExists(tableName, dbName) is True

# Does the column exist?
def test_columnExists():
  assert columnExists(df, columnName) is True

# Is there at least one row for the value in the specified column?
def test_numRowsInColumnForValue():
  assert numRowsInColumnForValue(df, columnName, columnValue) > 0

Exécuter les tests unitaires​

Cette section décrit comment exécuter les tests unitaires que vous avez codés dans la section précédente. Lorsque vous exécutez les tests unitaires, vous obtenez des résultats indiquant quels tests unitaires ont réussi et lesquels ont échoué.

Si vous avez ajouté les tests unitaires de la section précédente à votre workspace Databricks, vous pouvez exécuter ces tests unitaires à partir de votre workspace. Vous pouvez exécuter ces tests unitaires soit manuellement, soit selon un calendrier.

Créez un notebook Python dans le même dossier que le fichier test_myfunctions.py précédent dans votre repo, et ajoutez le contenu suivant.

Dans la première cellule du nouveau notebook, ajoutez le code suivant, puis exécutez la cellule, qui appelle la commande magique %pip. Cette commande magique installe pytest.

Python
%pip install pytest

Dans la deuxième cellule, ajoutez le code suivant, puis exécutez la cellule. Les résultats indiquent quels tests unitaires ont réussi et ont échoué.

Python
import pytest
import sys

# Skip writing pyc files on a readonly filesystem.
sys.dont_write_bytecode = True

# Run pytest.
retcode = pytest.main([".", "-v", "-p", "no:cacheprovider"])

# Fail the cell execution if there are any test failures.
assert retcode == 0, "The pytest invocation failed. See the log for details."
astuce

Vous pouvez afficher les résultats de vos exécutions de notebook (y compris les résultats des tests unitaires) dans les logs du Driver de votre cluster. Vous pouvez également spécifier un emplacement pour la livraison des Logs de votre cluster.

Vous pouvez configurer un système de fonctionnalités d’intégration et de livraison continues (CI/CD), tel que GitHub Actions, pour exécuter automatiquement vos tests unitaires chaque fois que votre code change. Pour un exemple, consultez la couverture de GitHub Actions dans Bonnes pratiques d'ingénierie logicielle pour les notebooks Databricks.

Ressources supplémentaires​

pytest​

testthat​

ScalaTest​

SQL​

Sur cette page