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

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

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

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

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

  • 関数とその単体テストを整理する方法。
  • 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 ワークスペースに追加できます。

次のコードは、 Databricks Git フォルダーを設定し、 リポジトリを追加し、Databricks ワークスペースでリポジトリが開いていることを前提としています。

リポジトリ内に myfunctions.py という名前のファイルを作成し、次の内容をファイルに追加します。この記事の他の例では、このファイルの名前が myfunctions.pyであると想定しています。独自のファイルに異なる名前を使用できます。

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 ワークスペースに追加した場合は、次のようにワークスペースからこれらの関数を呼び出すことができます。

リポジトリ内の前の myfunctions.py ファイルと同じフォルダーに 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}'.")

単体テストの記述​

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

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

リポジトリ内の前の myfunctions.py ファイルと同じフォルダーに test_myfunctions.py という名前の別のファイルを作成し、次の内容をファイルに追加します。デフォルトでは、pytest 名前が test_ で始まる (または _testで終わる) .py ファイルを検索してテストします。同様に、デフォルトで、 pytest はこれらのファイルの内部で、名前がテストする test_ で始まる関数を検索します。

一般に、本番運用でデータを操作する関数に対してユニットテストを実行し ない のがベストプラクティスです。 これは、データを追加、削除、または変更する関数にとって特に重要です。ユニットテストによって予期せぬ形で本番運用データが損なわれるのを防ぐために、ユニット運用以外のデータに対してユニット運用を実施する必要があります。 一般的なアプローチの1つは、本番運用データにできるだけ近い偽のデータを作成することです。 次のコード例では、単体テストの実行対象となる偽のデータを作成します。

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 ワークスペースに追加した場合は、ワークスペースからこれらの単体テストを実行できます。 これらの単体テストは、 手動で 実行することも、 スケジュールに従って実行することもできます。

リポジトリ内の前の test_myfunctions.py ファイルと同じフォルダーに Python ノートブックを作成し、次の内容を追加します。

新しいノートブックの最初のセルに次のコードを追加し、 %pip マジックを呼び出すセルを実行します。この魔法は pytestをインストールします。

%pip install pytest

2 番目のセルに次のコードを追加し、セルを実行します。結果には、成功した単体テストと失敗した単体テストが表示されます。

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の​

に最終更新