참고
이 문서에서는 Databricks Runtime 13.3 LTS 이상에 대한 Databricks Connect에 대해 설명합니다.
이 문서에서는 Databricks Runtime 13.3 LTS 이상용 Databricks Connect를 사용하여 pytest 테스트를 실행하는 방법을 설명합니다. Python용 Databricks Connect를 설치하려면 Python용 Databricks Connect 설치 참조하세요.
시작하려면 pytest의 설명서에서 시작하기를 pytest 참조하세요.
참고
터미널에서 Databricks Connect를 실행하는 경우 pytest DEFAULT 구성 프로필에서만 작동합니다. 프로필에는 클러스터 또는 서버리스 컴퓨팅 중에서 사용하려는 Databricks 컴퓨팅이 포함되어야 합니다. 컴퓨팅 구성에 대한 자세한 내용은 Databricks Connect 대한Compute 구성을 참조하세요.
예를 들어 인스턴스를 반환 nyctaxi_functions.py 하는 get_spark 함수와 SparkSession 카탈로그 get_nyctaxi_trips 스키마의 테이블을 DataFrame 나타내는 trips 함수 samples 를 포함하는 다음 nyctaxi 파일을 지정합니다.
nyctaxi_functions.py:
from databricks.connect import DatabricksSession
from pyspark.sql import DataFrame, SparkSession
def get_spark() -> SparkSession:
spark = DatabricksSession.builder.getOrCreate()
return spark
def get_nyctaxi_trips() -> DataFrame:
spark = get_spark()
df = spark.read.table("samples.nyctaxi.trips")
return df
main.py라는 다음 파일이 get_spark 및 get_nyctaxi_trips 함수를 호출합니다.
main.py:
from nyctaxi_functions import *
df = get_nyctaxi_trips()
df.show(5)
test_nyctaxi_functions.py라는 이름의 다음 파일은 get_spark 함수가 SparkSession 인스턴스를 반환하는지와 get_nyctaxi_trips 함수가 하나 이상의 데이터 행을 포함하는 DataFrame를 반환하는지를 테스트합니다.
test_nyctaxi_functions.py:
import pyspark.sql.connect.session
from nyctaxi_functions import *
def test_get_spark():
spark = get_spark()
assert isinstance(spark, pyspark.sql.connect.session.SparkSession)
def test_get_nyctaxi_trips():
df = get_nyctaxi_trips()
assert df.count() > 0
이러한 테스트를 실행하려면 코드 프로젝트의 루트에서 명령을 실행 pytest 합니다. 이 명령은 다음과 유사한 테스트 결과를 생성해야 합니다.
$ pytest
=================== test session starts ====================
platform darwin -- Python 3.11.7, pytest-8.1.1, pluggy-1.4.0
rootdir: <project-rootdir>
collected 2 items
test_nyctaxi_functions.py .. [100%]
======================== 2 passed ==========================