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.

O código a seguir pressupõe que o senhor tenha configurado as pastas Databricks Git, adicionado um repositório e que o repositório esteja aberto em seu Databricks workspace.

Crie um arquivo chamado myfunctions.py no repositório e adicione o seguinte conteúdo ao arquivo. Outros exemplos neste artigo esperam que esse arquivo tenha o nome myfunctions.py. Você pode usar nomes diferentes para seus próprios arquivos.

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.

Crie um Notebook Python na mesma pasta que o arquivo myfunctions.py anterior em seu repositório e adicione o seguinte conteúdo ao Notebook. Altere os valores das variáveis para o nome da tabela, o nome do esquema (banco de dados), o nome da coluna e o valor da coluna conforme necessário. Em seguida, anexe o Notebook a um cluster e execute o Notebook para ver os resultados.

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.

Crie outro arquivo chamado test_myfunctions.py na mesma pasta do arquivo myfunctions.py anterior em seu repositório e adicione o conteúdo a seguir ao arquivo. Por default, pytest procura por arquivos .py cujos nomes comecem com test_ (ou terminem com _test) para testar. Da mesma forma, por default, pytest procura dentro desses arquivos as funções cujos nomes começam com test_ para testar.

Em geral, é uma prática recomendada não executar testes de unidade em funções que trabalham com dados em produção. Isso é especialmente importante para funções que adicionam, removem ou alteram dados. Para evitar que os dados de produção sejam comprometidos pelos testes de unidade de forma inesperada, o senhor deve executar testes de unidade em dados que não sejam de produção. Uma abordagem comum é criar dados falsos que estejam o mais próximo possível dos dados de produção. O exemplo de código a seguir cria dados falsos para a execução dos testes de unidade.

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.

Crie um Notebook Python na mesma pasta que o arquivo test_myfunctions.py anterior em seu repositório e adicione o seguinte conteúdo.

Na primeira célula do novo Notebook, adicione o seguinte código e, em seguida, execute a célula que chama a mágica %pip. Essa mágica instala o pytest.

%pip install pytest

Na segunda célula, adicione o seguinte código e, em seguida, execute a célula. Os resultados mostram quais testes unitários foram aprovados e falharam.

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