メインコンテンツまでスキップ

ノートブックの単体テスト

単体テスト を使用すると、ノートブックのコードの品質と一貫性を向上させることができます。単体テストは、関数などの自己完結型のコード単位を早期かつ頻繁にテストするアプローチです。 これにより、コードの問題をより迅速に発見し、コードに関する誤った仮定を早期に発見し、全体的なコーディング作業を効率化できます。

この記事では、関数を使用した基本的な 単体テスト の概要です。 単体テストのクラスやインターフェイス、 スタブ、 モック、 テスト ハーネスの使用などの高度な概念は、ノートブックの単体テストでもサポートされていますが、この記事では説明しません。 この記事では、統合テスト 、システムテスト 、受け入れテスト 、または パフォーマンステスト やユーザビリティテスト などの非機能テスト 方法など、他の種類のテスト方法については説明しません。

この記事では、以下のトピックを紹介します。

  • 関数とその単体テストを整理する方法。
  • Python、R、Scala での関数の書き方、および SQL でのユーザー定義関数の書き方で、単体テストに適した設計になっています。
  • Python、R、Scala、SQL ノートブックからこれらの関数を呼び出す方法。
  • Python、R、Scala の一般的なテスト フレームワークである pytest for Python、 testthat 、および Scala for Scala を使用して単体テストを記述する方法。 また、ユニット テストの SQL ユーザー定義関数 (SQL UDF) の SQL の記述方法も説明します。
  • これらの単体テストを Python、R、Scala、SQL ノートブックから実行する方法。
注記

Databricks では、単体テストをノートブックに記述して実行することをお勧めします。 Webターミナルで一部のコマンドを実行できますが、WebターミナルにはSparkのサポートがないなど、より多くの制限があります。 「Databricks Webターミナルの実行 シェル コマンド」を参照してください。

関数と単体テストの整理​

ノートブックを使用して関数とその単体テストを整理するための一般的な方法がいくつかあります。 それぞれのアプローチには、それぞれの利点と課題があります。

Python、R、Scala ノートブックの場合、一般的なアプローチは次のとおりです。

  • 関数とその単体テストをノートブックの外部に格納します。

    • 利点: これらの関数は、ノートブックの内外で呼び出すことができます。 テスト フレームワークは、ノートブックの外部でテストを実行するように適切に設計されています。
    • 課題: このアプローチは Scala ノートブックではサポートされていません。 このアプローチでは、追跡および保守するファイルの数も増加します。

  • 関数を 1 つのノートブックに格納し、その単体テストを別のノートブックに格納します。

    • 利点: これらの関数は、ノートブック間で再利用しやすくなります。
    • 課題: 追跡および保守するノートブックの数が増加します。 これらの機能は、ノートブックの外部では使用できません。 また、これらの機能は、ノートブックの外部でテストするのがより困難になる場合があります。

  • 関数とその単体テストを同じノートブック内に格納します。

    • 利点: 関数とその単体テストは 1 つのノートブックに保存されるため、追跡と保守が容易になります。
    • 課題: これらの関数は、ノートブック間で再利用するのがより難しくなる可能性があります。 これらの機能は、ノートブックの外部では使用できません。 また、これらの機能は、ノートブックの外部でテストするのがより困難になる場合があります。

Python ノートブックと R ノートブックの場合、Databricks では関数とその単体テストをノートブックの外部に格納することをお勧めします。 Scalaノートブックの場合、Databricks では、関数を 1 つのノートブックに含め、その単体テストを別のノートブックに含めることをお勧めします。

SQL ノートブックの場合、Databricks では、関数を SQL ユーザー定義関数 (SQL UDF) としてスキーマ (データベースとも呼ばれます) に格納することをお勧めします。 その後、これらの SQL UDF とその単体テストを SQL ノートブックから呼び出すことができます。

書き込み関数​

このセクションでは、以下を決定する関数の簡単な例のセットについて説明します。

  • データベースにテーブルが存在するかどうか。
  • テーブルに列が存在するかどうか。
  • その列内の値に対して、列に存在する行数。

これらの関数は単純に作られているので、関数そのものに集中するよりも、この記事のユニットテストの詳細に集中できます。

最適な単体テスト結果を得るには、関数は 1 つの予測可能な結果を返し、1 つのデータ型である必要があります。 たとえば、何かが存在するかどうかを確認するには、関数は true または false のブール値を返す必要があります。 存在する行数を返すには、関数は負でない整数を返す必要があります。 最初の例では、何かが存在しない場合は false を返し、存在する場合はそれ自体を返すべきではありません。 同様に、2 番目の例では、存在する行の数を返さず、行が存在しない場合は false を返さないでください。

これらの関数は、Python、R、Scala、または SQL で次のように、既存の Databricks ワークスペースに追加できます。

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

関数を呼び出す​

このセクションでは、上記の関数を呼び出すコードについて説明します。 たとえば、これらの関数を使用して、指定した列内に指定した値が存在するテーブル内の行数をカウントできます。 ただし、先に進む前に、テーブルが実際に存在するかどうか、および列がそのテーブルに実際に存在するかどうかを確認する必要があります。 次のコードでは、これらの条件を確認します。

前のセクションの関数を Databricks ワークスペースに追加した場合は、次のようにワークスペースからこれらの関数を呼び出すことができます。

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

単体テストの記述​

このセクションでは、この記事の冒頭で説明した各関数をテストするコードについて説明します。 将来、関数に変更を加えた場合は、単体テストを使用して、それらの関数が期待どおりに動作するかどうかを判断できます。

この記事の冒頭で関数を Databricks ワークスペースに追加した場合は、次のようにして、これらの関数の単体テストをワークスペースに追加できます。

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

単体テストの実行​

このセクションでは、前のセクションでコーディングした単体テストを実行する方法について説明します。 単体テストを実行すると、成功した単体テストと失敗した単体テストの結果が表示されます。

前のセクションの単体テストを Databricks ワークスペースに追加した場合は、ワークスペースからこれらの単体テストを実行できます。 これらの単体テストは、 手動で 実行することも、 スケジュールに従って実行することもできます。

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."
ヒント

ノートブックの実行結果 (単体テストの結果を含む) は、クラスターの ドライバー ログで表示できます。 クラスターの ログ配信の場所を指定することもできます。

などの継続的インテグレーションと継続的デリバリーまたはデプロイメントCI/CD ()GitHub Actions システムを設定して、コードが変更されるたびにユニット・テストを自動的に実行できます。例については、「 ノートブックのソフトウェア エンジニアリングのベスト プラクティス」の GitHub Actions のカバレッジを参照してください。

追加のリソース​

pytestの​

テストを​

スカラテスト​

SQLの​

に最終更新