Partager via

Tests unitaires pour les notebooks

Vous pouvez utiliser des tests unitaires pour améliorer la qualité et la cohérence du code de vos notebooks. Le test unitaire est une approche pour tester des unités de code autonomes, telles que des fonctions, tôt dans le processus et souvent. Cela vous aide à trouver plus rapidement les problèmes sur votre code, à révéler plus tôt les prévisions erronées et à simplifier vos efforts de codage.

Cet article présente les tests unitaires de base avec les fonctions. Les concepts avancés tels que les classes et interfaces de test unitaire, ainsi que l’utilisation de stubs, de mocks et d’un environnement de test, qui sont pris en charge lors des tests unitaires pour les notebooks, ne sont pas abordés dans cet article. Cet article ne couvre pas non plus d’autres types de méthodes de test, telles que les tests d’intégration, lestests système, les tests d’acceptation ou les méthodes de test non fonctionnelles telles que les tests de performances ou les tests d’utilisation.

Cet article illustre les éléments suivants :

  • 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 dans 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 à l’aide des frameworks de test populaires pytest pour Python, testthat pour R et ScalaTest pour Scala . En outre, comment écrire du code SQL pour tester les fonctions SQL définies par l'utilisateur.
  • Comment exécuter ces tests unitaires à partir de notebooks Python, R, Scala et SQL.

Remarque

Azure Databricks recommande d’écrire et d’exécuter vos tests unitaires dans un notebook. Bien que vous puissiez exécuter certaines commandes dans le terminal web, le terminal web présente plus de limitations, telles qu’un manque de prise en charge de Spark. Consultez Exécuter des commandes d’interpréteur de commandes dans un terminal web Azure Databricks.

Organiser des fonctions et des tests unitaires

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

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

  • Stockez les fonctions et leurs tests unitaires en dehors des notebooks.
    • Avantages : vous pouvez appeler ces fonctions depuis les notebooks ou en dehors. Les frameworks de test sont mieux conçus pour exécuter des tests en dehors des notebooks.
    • Défis : cette approche n’est pas prise en charge pour les notebooks Scala. Cette approche augmente également le nombre de fichiers à suivre et à gérer.
  • Stockez des fonctions dans un bloc-notes et leurs tests unitaires dans un bloc-notes distinct.
    • Avantages : ces fonctions sont plus faciles à réutiliser dans les notebooks.
    • Défis : Le nombre de blocs-notes à 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 notebooks.
  • 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 dans 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 notebooks.

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 des fonctions dans un bloc-notes et leurs tests unitaires dans un bloc-notes distinct.

Pour les notebooks SQL, Databricks vous recommande de stocker des fonctions en tant que fonctions SQL définies par l’utilisateur (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.

Écrire des fonctions

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.
  • Nombre de lignes dans une colonne pour une valeur dans cette colonne.

Ces fonctions sont destinées à être simples, afin que vous puissiez vous concentrer sur les détails des tests unitaires dans cet article plutôt que sur les fonctions elles-mêmes.

Pour obtenir les meilleurs résultats de tests unitaires, une fonction doit retourner un résultat prévisible unique et être d’un type de données unique. Par exemple, pour vérifier si quelque chose existe, la fonction doit retourner une valeur booléenne true ou false. Pour retourner le nombre de lignes qui existent, la fonction doit retourner un nombre entier non négatif. Dans le premier exemple, il ne doit ni retourner false si quelque chose n'existe pas, ni retourner la chose elle-même si elle existe. De même, pour le deuxième exemple, il ne doit pas retourner le nombre de lignes qui existent ou false si aucune ligne n’existe.

Vous pouvez ajouter ces fonctions à un espace de travail Azure Databricks existant comme suit, en Python, R, Scala ou SQL.

Python

Le code suivant part du principe que vous avez configuré des dossiers Git Databricks, ajouté un dépôt et que le dépôt est ouvert dans votre espace de travail Azure Databricks.

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

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()

R

Le code suivant part du principe que vous avez configuré des dossiers Git Databricks, ajouté un dépôt et que le dépôt est ouvert dans votre espace de travail Azure Databricks.

Créez un fichier nommé myfunctions.r dans le dépôt et ajoutez le contenu suivant au fichier. D’autres exemples de cet article s’attendent à ce que ce fichier soit nommé myfunctions.r. Vous pouvez utiliser différents noms pour vos propres fichiers.

library(SparkR)

# Does the specified table exist in the specified database?
table_exists <- function(table_name, db_name) {
  tableExists(paste(db_name, ".", table_name, sep = ""))
}

# Does the specified column exist in the given DataFrame?
column_exists <- function(dataframe, column_name) {
  column_name %in% colnames(dataframe)
}

# How many rows are there for the specified value in the specified column
# in the given DataFrame?
num_rows_in_column_for_value <- function(dataframe, column_name, column_value) {
  df = filter(dataframe, dataframe[[column_name]] == column_value)

  count(df)
}

Langage de programmation Scala

Créez un notebook Scala nommé myfunctions avec le contenu suivant. D’autres exemples de cet article supposent que ce bloc-notes soit nommé myfunctions. Vous pouvez utiliser différents noms pour vos propres blocs-notes.

import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.functions.col

// Does the specified table exist in the specified database?
def tableExists(tableName: String, dbName: String) : Boolean = {
  return spark.catalog.tableExists(dbName + "." + tableName)
}

// Does the specified column exist in the given DataFrame?
def columnExists(dataFrame: DataFrame, columnName: String) : Boolean = {
  val nameOfColumn = null

  for(nameOfColumn <- dataFrame.columns) {
    if (nameOfColumn == columnName) {
      return true
    }
  }

  return false
}

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

  return df.count()
}

SQL

Le code suivant suppose que vous disposez de l’exemple de jeu de données tiers diamonds au sein d’un schéma nommé default dans un catalogue nommé main accessible à partir de votre espace de travail Azure Databricks. Si le catalogue ou le schéma que vous souhaitez utiliser a un nom différent, modifiez une ou les deux instructions suivantes USE pour qu’elles correspondent.

Créez un bloc-notes SQL et ajoutez le contenu suivant à ce nouveau notebook. Attachez ensuite le notebook à un cluster et exécutez le notebook pour ajouter les fonctions définies par l'utilisateur SQL suivantes au catalogue et au schéma spécifiés.

Remarque

Les fonctions définies par l'utilisateur SQL table_exists et column_exists fonctionnent uniquement avec Unity Catalog. La prise en charge des fonctions UDF SQL pour le catalogue Unity est en préversion publique.

USE CATALOG main;
USE SCHEMA default;

CREATE OR REPLACE FUNCTION table_exists(catalog_name STRING,
                                        db_name      STRING,
                                        table_name   STRING)
  RETURNS BOOLEAN
  RETURN if(
    (SELECT count(*) FROM system.information_schema.tables
     WHERE table_catalog = table_exists.catalog_name
       AND table_schema  = table_exists.db_name
       AND table_name    = table_exists.table_name) > 0,
    true,
    false
  );

CREATE OR REPLACE FUNCTION column_exists(catalog_name STRING,
                                         db_name      STRING,
                                         table_name   STRING,
                                         column_name  STRING)
  RETURNS BOOLEAN
  RETURN if(
    (SELECT count(*) FROM system.information_schema.columns
     WHERE table_catalog = column_exists.catalog_name
       AND table_schema  = column_exists.db_name
       AND table_name    = column_exists.table_name
       AND column_name   = column_exists.column_name) > 0,
    true,
    false
  );

CREATE OR REPLACE FUNCTION num_rows_for_clarity_in_diamonds(clarity_value STRING)
  RETURNS BIGINT
  RETURN SELECT count(*)
         FROM main.default.diamonds
         WHERE clarity = clarity_value

Fonctions d’appel

Cette section décrit le code qui appelle les fonctions précédentes. Vous pouvez utiliser ces fonctions, par exemple, pour compter le nombre de lignes dans la table où une valeur spécifiée existe dans une colonne spécifiée. Toutefois, vous souhaitez 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 espace de travail Azure Databricks, vous pouvez appeler ces fonctions à partir de votre espace de travail comme suit.

Python

Créez un bloc-notes Python dans le même dossier que le fichier précédent myfunctions.py de votre dépôt, puis ajoutez le contenu suivant au bloc-notes. Modifiez les valeurs de variable pour le nom de la table, le nom du schéma (base de données), le nom de colonne et la valeur de colonne si nécessaire. Attachez ensuite le bloc-notes à un cluster et exécutez le notebook pour afficher les résultats.

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}'.")

R

Créez un bloc-notes R dans le même dossier que le fichier précédent myfunctions.r de votre dépôt, puis ajoutez le contenu suivant au bloc-notes. Modifiez les valeurs de variable pour le nom de la table, le nom du schéma (base de données), le nom de colonne et la valeur de colonne si nécessaire. Attachez ensuite le bloc-notes à un cluster et exécutez le notebook pour afficher les résultats.

library(SparkR)
source("myfunctions.r")

table_name   <- "diamonds"
db_name      <- "default"
column_name  <- "clarity"
column_value <- "VVS2"

# If the table exists in the specified database...
if (table_exists(table_name, db_name)) {

  df = sql(paste("SELECT * FROM ", db_name, ".", table_name, sep = ""))

  # And the specified column exists in that table...
  if (column_exists(df, column_name)) {
    # Then report the number of rows for the specified value in that column.
    num_rows = num_rows_in_column_for_value(df, column_name, column_value)

    print(paste("There are ", num_rows, " rows in table '", table_name, "' where '", column_name, "' equals '", column_value, "'.", sep = "")) 
  } else {
    print(paste("Column '", column_name, "' does not exist in table '", table_name, "' in schema (database) '", db_name, "'.", sep = ""))
  }

} else {
  print(paste("Table '", table_name, "' does not exist in schema (database) '", db_name, "'.", sep = ""))
}

Langage de programmation Scala

Créez un autre bloc-notes Scala dans le même dossier que le bloc-notes Scala précédent myfunctions et ajoutez le contenu suivant à ce nouveau notebook.

Dans la première cellule de ce nouveau notebook, ajoutez le code suivant, qui utilise le "magique" %run. Cette magie rend le contenu du myfunctions bloc-notes disponible pour votre nouveau bloc-notes.

%run ./myfunctions

Dans la deuxième cellule de ce nouveau bloc-notes, ajoutez le code suivant. Modifiez les valeurs de variable pour le nom de la table, le nom du schéma (base de données), le nom de colonne et la valeur de colonne si nécessaire. Attachez ensuite le bloc-notes à un cluster et exécutez le notebook pour afficher les résultats.

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

// If the table exists in the specified database...
if (tableExists(tableName, dbName)) {

  val df = spark.sql("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.
    val numRows = numRowsInColumnForValue(df, columnName, columnValue)

    println("There are " + numRows + " rows in '" + tableName + "' where '" + columnName + "' equals '" + columnValue + "'.")
  } else {
    println("Column '" + columnName + "' does not exist in table '" + tableName + "' in database '" + dbName + "'.")
  }

} else {
  println("Table '" + tableName + "' does not exist in database '" + dbName + "'.")
}

SQL

Ajoutez le code suivant à une nouvelle cellule du bloc-notes précédent ou à une cellule d’un bloc-notes distinct. Modifiez le schéma ou les noms de catalogue si nécessaire pour correspondre à la vôtre, puis exécutez cette cellule pour afficher les résultats.

SELECT CASE
-- If the table exists in the specified catalog and schema...
WHEN
  table_exists("main", "default", "diamonds")
THEN
  -- And the specified column exists in that table...
  (SELECT CASE
   WHEN
     column_exists("main", "default", "diamonds", "clarity")
   THEN
     -- Then report the number of rows for the specified value in that column.
     printf("There are %d rows in table 'main.default.diamonds' where 'clarity' equals 'VVS2'.",
            num_rows_for_clarity_in_diamonds("VVS2"))
   ELSE
     printf("Column 'clarity' does not exist in table 'main.default.diamonds'.")
   END)
ELSE
  printf("Table 'main.default.diamonds' does not exist.")
END

Écrire des tests unitaires

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

Si vous avez ajouté les fonctions vers le début de cet article à votre espace de travail Azure Databricks, vous pouvez ajouter des tests unitaires pour ces fonctions à votre espace de travail comme suit.

Python

Créez un autre fichier nommé test_myfunctions.py dans le même dossier que le fichier précédent myfunctions.py dans votre dépôt, puis ajoutez le contenu suivant au fichier. Par défaut, pytest recherche les .py fichiers dont les noms commencent par test_ (ou se terminent par _test) à tester. De même, par défaut, pytest examine l'intérieur de ces fichiers pour rechercher des fonctions dont les noms commencent par test_ afin de les tester.

En général, il est recommandé de ne pas exécuter de tests unitaires sur des fonctions qui fonctionnent avec des données en production. Cela est particulièrement important pour les fonctions qui ajoutent, suppriment ou modifient des données. Pour protéger vos données de production contre la compromission par vos tests unitaires de manière inattendue, vous devez exécuter des tests unitaires sur des données hors production. Une approche courante consiste à créer des données factices aussi proches que possible des données de production. L'exemple de code suivant crée des données factices pour tester les tests unitaires.

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

R

Créez un autre fichier nommé test_myfunctions.r dans le même dossier que le fichier précédent myfunctions.r dans votre dépôt, puis ajoutez le contenu suivant au fichier. Par défaut, testthat recherche les fichiers .r dont les noms commencent par test pour les tester.

En général, il est recommandé de ne pas exécuter de tests unitaires sur des fonctions qui fonctionnent avec des données en production. Cela est particulièrement important pour les fonctions qui ajoutent, suppriment ou modifient des données. Pour protéger vos données de production contre la compromission par vos tests unitaires de manière inattendue, vous devez exécuter des tests unitaires sur des données hors production. Une approche courante consiste à créer des données factices aussi proches que possible des données de production. L'exemple de code suivant crée des données factices pour tester les tests unitaires.

library(testthat)
source("myfunctions.r")

table_name   <- "diamonds"
db_name      <- "default"
column_name  <- "clarity"
column_value <- "SI2"

# 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",     "integer"),
  structField("carat",   "float"),
  structField("cut",     "string"),
  structField("color",   "string"),
  structField("clarity", "string"),
  structField("depth",   "float"),
  structField("table",   "integer"),
  structField("price",   "integer"),
  structField("x",       "float"),
  structField("y",       "float"),
  structField("z",       "float"))

data <- list(list(as.integer(1), 0.23, "Ideal",   "E", "SI2", 61.5, as.integer(55), as.integer(326), 3.95, 3.98, 2.43),
             list(as.integer(2), 0.21, "Premium", "E", "SI1", 59.8, as.integer(61), as.integer(326), 3.89, 3.84, 2.31))

df <- createDataFrame(data, schema)

# Does the table exist?
test_that ("The table exists.", {
  expect_true(table_exists(table_name, db_name))
})

# Does the column exist?
test_that ("The column exists in the table.", {
  expect_true(column_exists(df, column_name))
})

# Is there at least one row for the value in the specified column?
test_that ("There is at least one row in the query result.", {
  expect_true(num_rows_in_column_for_value(df, column_name, column_value) > 0)
})

Langage de programmation Scala

Créez un autre bloc-notes Scala dans le même dossier que le bloc-notes Scala précédent myfunctions et ajoutez le contenu suivant à ce nouveau notebook.

Dans la première cellule du nouveau notebook, ajoutez le code suivant, qui utilise la commande %run. Cette magie rend le contenu du myfunctions bloc-notes disponible pour votre nouveau bloc-notes.

%run ./myfunctions

Dans la deuxième cellule, ajoutez le code suivant. Ce code définit vos tests unitaires et spécifie comment les exécuter.

En général, il est recommandé de ne pas exécuter de tests unitaires sur des fonctions qui fonctionnent avec des données en production. Cela est particulièrement important pour les fonctions qui ajoutent, suppriment ou modifient des données. Pour protéger vos données de production contre la compromission par vos tests unitaires de manière inattendue, vous devez exécuter des tests unitaires sur des données hors production. Une approche courante consiste à créer des données factices aussi proches que possible des données de production. L'exemple de code suivant crée des données factices pour tester les tests unitaires.

import org.scalatest._
import org.apache.spark.sql.types.{StructType, StructField, IntegerType, FloatType, StringType}
import scala.collection.JavaConverters._

class DataTests extends AsyncFunSuite {

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

  // 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.
  val schema = StructType(Array(
                 StructField("_c0",     IntegerType),
                 StructField("carat",   FloatType),
                 StructField("cut",     StringType),
                 StructField("color",   StringType),
                 StructField("clarity", StringType),
                 StructField("depth",   FloatType),
                 StructField("table",   IntegerType),
                 StructField("price",   IntegerType),
                 StructField("x",       FloatType),
                 StructField("y",       FloatType),
                 StructField("z",       FloatType)
               ))

  val data = Seq(
                  Row(1, 0.23, "Ideal",   "E", "SI2", 61.5, 55, 326, 3.95, 3.98, 2.43),
                  Row(2, 0.21, "Premium", "E", "SI1", 59.8, 61, 326, 3.89, 3.84, 2.31)
                ).asJava

  val df = spark.createDataFrame(data, schema)

  // Does the table exist?
  test("The table exists") {
    assert(tableExists(tableName, dbName) == true)
  }

  // Does the column exist?
  test("The column exists") {
    assert(columnExists(df, columnName) == true)
  }

  // Is there at least one row for the value in the specified column?
  test("There is at least one matching row") {
    assert(numRowsInColumnForValue(df, columnName, columnValue) > 0)
  }
}

nocolor.nodurations.nostacks.stats.run(new DataTests)

Remarque

Cet exemple de code utilise le FunSuite style de test dans ScalaTest. Pour d’autres styles de test disponibles, consultez Sélection de styles de test pour votre projet.

SQL

Avant d’ajouter des tests unitaires, vous devez savoir qu’en général, il est recommandé de ne pas exécuter de tests unitaires sur des fonctions qui fonctionnent avec des données en production. Cela est particulièrement important pour les fonctions qui ajoutent, suppriment ou modifient des données. Pour protéger vos données de production contre la compromission par vos tests unitaires de manière inattendue, vous devez exécuter des tests unitaires sur des données hors production. Une approche courante consiste à exécuter des tests unitaires sur des vues plutôt que sur des tables.

Pour créer un affichage, vous pouvez appeler la CREATE VIEW commande à partir d’une nouvelle cellule dans le bloc-notes précédent ou un bloc-notes distinct. L’exemple suivant suppose que vous disposez d’une table existante nommée diamonds dans un schéma (base de données) nommé default dans un catalogue nommé main. Modifiez ces noms pour qu’ils correspondent à vos propres besoins, puis exécutez uniquement cette cellule.

USE CATALOG main;
USE SCHEMA default;

CREATE VIEW view_diamonds AS
SELECT * FROM diamonds;

Après avoir créé la vue, ajoutez chaque instruction suivante SELECT à sa propre cellule dans le bloc-notes précédent ou dans un bloc-notes distinct. Modifiez les noms pour qu’ils correspondent à vos propres besoins.

SELECT if(table_exists("main", "default", "view_diamonds"),
          printf("PASS: The table 'main.default.view_diamonds' exists."),
          printf("FAIL: The table 'main.default.view_diamonds' does not exist."));

SELECT if(column_exists("main", "default", "view_diamonds", "clarity"),
          printf("PASS: The column 'clarity' exists in the table 'main.default.view_diamonds'."),
          printf("FAIL: The column 'clarity' does not exists in the table 'main.default.view_diamonds'."));

SELECT if(num_rows_for_clarity_in_diamonds("VVS2") > 0,
          printf("PASS: The table 'main.default.view_diamonds' has at least one row where the column 'clarity' equals 'VVS2'."),
          printf("FAIL: The table 'main.default.view_diamonds' does not have at least one row where the column 'clarity' equals 'VVS2'."));

Exécuter des tests unitaires

Cette section explique 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 échoué.

Si vous avez ajouté les tests unitaires de la section précédente à votre espace de travail Azure Databricks, vous pouvez exécuter ces tests unitaires à partir de votre espace de travail. Vous pouvez exécuter ces tests unitaires manuellement ou selon une planification.

Python

Créez un notebook Python dans le même dossier que le fichier précédent test_myfunctions.py de votre dépôt, puis ajoutez le contenu suivant.

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

%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 échoué.

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."

R

Créez un bloc-notes R dans le même dossier que le fichier précédent test_myfunctions.r de votre dépôt, puis ajoutez le contenu suivant.

Dans la première cellule, ajoutez le code suivant, puis exécutez la cellule, qui appelle la install.packages fonction. Cette fonction installe testthat.

install.packages("testthat")

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 échoué.

library(testthat)
source("myfunctions.r")

test_dir(".", reporter = "tap")

Langage de programmation Scala

Exécutez d'abord la première cellule, puis la deuxième dans le bloc-notes de la section précédente. Les résultats indiquent quels tests unitaires ont réussi et échoué.

SQL

Exécutez chacune des trois cellules du bloc-notes à partir de la section précédente. Les résultats indiquent si chaque test unitaire a réussi ou échoué.

Si vous n’avez plus besoin de la vue après avoir exécuté vos tests unitaires, vous pouvez supprimer la vue. Pour supprimer cette vue, vous pouvez ajouter le code suivant à une nouvelle cellule dans l’un des blocs-notes précédents, puis exécuter uniquement cette cellule.

DROP VIEW view_diamonds;

Conseil / Astuce

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

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

Ressources supplémentaires

pytest

testthat

ScalaTest

SQL