A Pythonhoz készült Databricks Connect tesztelése

Megjegyzés

Ez a cikk a Databricks Runtime 13.3 LTS-hez készült Databricks Connectet ismerteti.

Ez a cikk azt ismerteti, hogyan futtathat teszteket a Databricks Connect for Databricks Runtime 13.3 LTS és újabb verziók használatával pytest . A Databricks Connect for Python telepítéséhez lásd: Databricks Connect for Python telepítése.

Első lépésként tekintse meg a pytest dokumentáció Első lépések részét pytest.

Fontos

A Databricks Connect és a PySpark kölcsönösen kizárják egymást. További információ: Ütköző PySpark-telepítések.

Megjegyzés

Amikor a Databricks Connectet a terminálról futtatja, pytest csak az ALAPÉRTELMEZETT konfigurációs profillal működik. A profilnak tartalmaznia kell a használni kívánt Databricks-számítást, akár fürtöt, akár kiszolgáló nélküli számítást. A számítás konfigurálásával kapcsolatos információkért lásd a Databricks Connect számítási konfigurációját.

Ha például a következő fájl neve nyctaxi_functions.py egy get_spark olyan függvényt tartalmaz, amely egy példányt SparkSession ad vissza, és egy get_nyctaxi_trips olyan függvényt, amely a DataFrame katalógus sémájában trips szereplő samples táblát adja visszanyctaxi:

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

Adott a következő, main.py nevű fájl, amely hívja ezeket a get_spark és get_nyctaxi_trips függvényeket:

main.py:

from nyctaxi_functions import *

df = get_nyctaxi_trips()
df.show(5)

A következő elnevezett test_nyctaxi_functions.py fájl ellenőrzi, hogy a get_spark függvény egy példányt SparkSession ad-e vissza, és hogy a get_nyctaxi_trips függvény legalább egy adatsort tartalmazó értéket ad-e vissza 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

A tesztek futtatásához futtassa a parancsot a pytest kódprojekt gyökérkönyvtárából, amelynek a következőhöz hasonló teszteredményeket kell létrehoznia:

$ 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 ==========================