Pular para o conteúdo principal

Teste de unidade para o Notebook

O senhor pode usar o teste de unidade para ajudar a melhorar a qualidade e a consistência do código do Notebook. O teste unitário é uma abordagem para testar unidades de código independentes, como funções, com antecedência e frequência. Isso ajuda você a encontrar problemas com seu código com mais rapidez, descobrir suposições errôneas sobre seu código mais cedo e otimizar seus esforços gerais de codificação.

Este artigo é uma introdução aos testes unitários básicos com funções. Conceitos avançados, como classes e interfaces de teste de unidade, bem como o uso de stubs, mocks e chicotes de teste, embora também sejam compatíveis com o teste de unidade do Notebook, estão fora do escopo deste artigo. Este artigo também não abrange outros tipos de métodos de teste, como teste de integração, teste de sistema, teste de aceitação ou métodos de teste não funcionais, como teste de desempenho ou teste de usabilidade.

Este artigo demonstra o seguinte:

  • Como organizar funções e seus testes unitários.
  • Como escrever funções em Python, R, Scala, bem como funções definidas pelo usuário em SQL, que são bem projetadas para serem testadas por unidades.
  • Como chamar essas funções em Python, R, Scala e SQL Notebook.
  • Como escrever testes unitários em Python, R e Scala usando as estruturas de teste populares pytest para Python, testthat para R e ScalaTest para Scala. Além disso, como escrever SQL para testes unitários de funções definidas pelo usuário de SQL (SQL UDFs).
  • Como executar esses testes de unidade em Python, R, Scala e SQL Notebook.
nota

Databricks recomenda escrever e executar seus testes de unidade em um Notebook. Embora o senhor possa executar alguns comandos no terminal da Web, o terminal da Web tem mais limitações, como a falta de suporte para Spark. Consulte execução shell comando em Databricks web terminal.

Organize funções e testes unitários​

Há algumas abordagens comuns para organizar suas funções e seus testes de unidade com o Notebook. Cada abordagem tem seus benefícios e desafios.

Para Python, R e Scala Notebook, as abordagens comuns incluem o seguinte:

  • Armazenar funções e seus testes de unidade fora do Notebook.

    • Benefícios: O senhor pode chamar essas funções com e fora do Notebook. As estruturas de teste são mais bem projetadas para executar testes fora do Notebook.
    • Desafios: Essa abordagem não é compatível com o site Scala Notebook. Essa abordagem também aumenta o número de arquivos a serem rastreados e mantidos.

  • Armazene funções em um Notebook e seus testes de unidade em um Notebook separado.

    • Vantagens: Essas funções são mais fáceis de reutilizar no Notebook.
    • Desafios: O número de notebooks a serem monitorados e mantidos aumenta. Essas funções não podem ser usadas fora do Notebook. Essas funções também podem ser mais difíceis de testar fora do Notebook.

  • Armazenar funções e seus testes de unidade no mesmo Notebook.

    • Benefícios: As funções e seus testes unitários são armazenados em um único Notebook para facilitar o acompanhamento e a manutenção.
    • Desafios: Essas funções podem ser mais difíceis de reutilizar no Notebook. Essas funções não podem ser usadas fora do Notebook. Essas funções também podem ser mais difíceis de testar fora do Notebook.

Para o Python e o R Notebook, o Databricks recomenda armazenar funções e seus testes de unidade fora do Notebook. Para o Scala Notebook, o Databricks recomenda incluir funções em um Notebook e seus testes de unidade em um Notebook separado.

Para o SQL Notebook, o Databricks recomenda que o senhor armazene funções como SQL funções definidas pelo usuário (SQL UDFs) em seus esquemas (também conhecidos como bancos de dados). Em seguida, o senhor pode chamar esses UDFs SQL e seus testes unitários no SQL Notebook.

Funções de escrita​

Esta seção descreve um conjunto simples de exemplos de funções que determinam o seguinte:

  • Se existe uma tabela em um banco de dados.
  • Se existe uma coluna em uma tabela.
  • Quantas linhas existem em uma coluna para um valor dentro dessa coluna.

A intenção é que essas funções sejam simples, para que o senhor possa se concentrar nos detalhes do teste de unidade neste artigo, em vez de se concentrar nas próprias funções.

Para obter os melhores resultados de testes unitários, uma função deve retornar um único resultado previsível e ter um único tipo de dados. Por exemplo, para verificar se algo existe, a função deve retornar um valor booleano de verdadeiro ou falso. Para retornar o número de linhas existentes, a função deve retornar um número inteiro não negativo. No primeiro exemplo, não deve retornar nem falso se algo não existir, nem a coisa em si, se existir. Da mesma forma, para o segundo exemplo, ele não deve retornar o número de linhas existentes ou falso se não houver nenhuma linha.

O senhor pode adicionar essas funções a um Databricks workspace existente da seguinte forma, em Python, R, Scala, ou SQL.

The following code assumes you have Set up Databricks Git folders (Repos), added a repo, and have the repo open in your Databricks workspace.

Create a file named myfunctions.py within the repo, and add the following contents to the file. Other examples in this article expect this file to be named myfunctions.py. You can use different names for your own files.

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

Funções de chamada​

Esta seção descreve o código que chama as funções anteriores. Você pode usar essas funções, por exemplo, para contar o número de linhas na tabela em que um valor especificado existe em uma coluna especificada. No entanto, você gostaria de verificar se a tabela realmente existe e se a coluna realmente existe nessa tabela antes de continuar. O código a seguir verifica essas condições.

Se tiver adicionado as funções da seção anterior ao seu Databricks workspace, poderá chamar essas funções do seu workspace da seguinte forma.

Create a Python notebook in the same folder as the preceding myfunctions.py file in your repo, and add the following contents to the notebook. Change the variable values for the table name, the schema (database) name, the column name, and the column value as needed. Then attach the notebook to a cluster and run the notebook to see the results.

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

Escreva testes unitários​

Esta seção descreve o código que testa cada uma das funções descritas no início deste artigo. Se você fizer alguma alteração nas funções no futuro, poderá usar testes de unidade para determinar se essas funções ainda funcionam conforme o esperado.

Se o senhor adicionou as funções do início deste artigo ao seu Databricks workspace, poderá adicionar testes de unidade para essas funções ao seu workspace da seguinte forma.

Create another file named test_myfunctions.py in the same folder as the preceding myfunctions.py file in your repo, and add the following contents to the file. By default, pytest looks for .py files whose names start with test_ (or end with _test) to test. Similarly, by default, pytest looks inside of these files for functions whose names start with test_ to test.

In general, it is a best practice to not run unit tests against functions that work with data in production. This is especially important for functions that add, remove, or otherwise change data. To protect your production data from being compromised by your unit tests in unexpected ways, you should run unit tests against non-production data. One common approach is to create fake data that is as close as possible to the production data. The following code example creates fake data for the unit tests to run against.

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

execução de testes unitários​

Esta seção descreve como executar os testes unitários que o senhor codificou na seção anterior. Quando o senhor executa os testes de unidade, obtém resultados que mostram quais testes de unidade foram aprovados e quais falharam.

Se o senhor adicionou os testes de unidade da seção anterior ao site Databricks workspace, poderá executar esses testes de unidade no site workspace. O senhor pode executar esses testes unitários manualmente ou em um programador.

Create a Python notebook in the same folder as the preceding test_myfunctions.py file in your repo, and add the following contents.

In the new notebook’s first cell, add the following code, and then run the cell, which calls the %pip magic. This magic installs pytest.

%pip install pytest

In the second cell, add the following code and then run the cell. Results show which unit tests passed and failed.

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

O senhor logs pode view os resultados da execução do Notebook (incluindo os resultados dos testes de unidade) no driver do seu cluster. O senhor também pode especificar um local para a entrega log do do seu clustering.

O senhor pode configurar um sistema de integração contínua (CI) e entrega contínua (CD) ou de implantação (CI/CD), como o GitHub Actions, para executar automaticamente os testes de unidade sempre que o código for alterado. Como exemplo, veja a cobertura do site GitHub Actions nas práticas recomendadas de engenharia de software para o Notebook.

Recurso adicional​

pytest​

teste isso​

Teste de escala​

SQL​

Última atualização em