我需要使用 pytest 測驗我的 Spark 專案，但我不明白如何創建 Spark 會話。我做了一些研究并得出了以下結論：
import pytest
import unittest
from pyspark.sql import SparkSession
@pytest.fixture(scope="session")
def spark_session():
spark = SparkSession.builder.master("local[*]").appName("test").getOrCreate()
return spark
class Testdb2connection(unittest.TestCase):
@pytest.mark.usefixtures("spark_session")
def test_connectdb2(self):
with self.assertRaises(ValueError):
return spark_session.format('jdbc')\
...
但是，在運行測驗時，我得到：
'AttributeError:'function' 物件沒有屬性 'format'
我究竟做錯了什么？
uj5u.com熱心網友回復：
查看使用標記將 pytest 固定裝置混合到 unittest.TestCase 子類中，您可以定義spark_session范圍class并將火花會話添加到cls請求背景關系的屬性中，以便能夠將其用作使用該固定裝置的類中的屬性。
嘗試使用以下修改后的代碼：
import pytest
import unittest
from pyspark.sql import SparkSession
@pytest.fixture(scope='class')
def spark_session(request):
spark = SparkSession.builder.master("local[*]").appName("test").getOrCreate()
request.addfinalizer(lambda: spark.stop()) # to teardown the session after test
request.cls.spark = spark
@pytest.mark.usefixtures("spark_session")
class Testdb2connection(unittest.TestCase):
def test_connectdb2(self):
assert isinstance(self.spark, SparkSession)
運行測驗：
pytest ./mytest.py
. [100%]
1 passed in 4.12s
轉載請註明出處，本文鏈接：https://www.uj5u.com/gongcheng/394035.html