我有這個功能：
# spark already defined somewhere as:
spark = SparkSession.builder.appName("App").getOrCreate()
def read_data(spark):
query = "SELECT * FROM table"
pandas_df = pd.read_sql(query, conn)
return spark.createDataFrame(pandas_df)
要測驗它：
from unittest import mock
@mock.patch("pandas.read_sql")
@mock.patch("pyspark.sql.SparkSession", autospec=True)
def test_read_data(spark_session, pandas_read_sql):
result = read_data(spark_session)
assert == ???
我應該以什么方式測驗這個有意義的？任何幫助表示贊賞。
uj5u.com熱心網友回復：
為了測驗你的函式，你pandas.read_sql只需要模擬，spark_session不能被模擬，你需要有一個實體來正確測驗你的函式。您可以創建自己的pytest.fixture以滿足此要求。
from unittest.mock import patch
import pandas
import pyspark.sql
import pytest
from pyspark.sql import SparkSession
from your_module import read_data
@pytest.fixture
def spark_session():
_spark_session = SparkSession.builder.appName("unit-tests").getOrCreate()
yield _spark_session
_spark_session.stop()
@patch("pandas.read_sql")
def test_read_data(mock_read_sql, spark_session):
# given:
mock_read_sql.return_value = pandas.DataFrame(
[(1, "row1"), (2, "row2")], columns=["id", "column1"]
)
# when:
spark_df = read_data(spark_session)
# then:
assert isinstance(spark_df, pyspark.sql.DataFrame)
您可以做更多斷言并檢查創建的資料框是否具有正確的架構并包含您期望的值。
提示：您應該查看 spark sql 功能，因為您可能不需要使用 Pandas 來查詢您的資料庫。
轉載請註明出處，本文鏈接：https://www.uj5u.com/caozuo/344585.html
標籤：Python 熊猫 火花 python-unittest
上一篇：繪圖，按標簽，詞頻