Python unittest mock pyspark chain Python unittest mock permite simular el comportamiento de objetos y funciones en las pruebas unitarias de Python. El módulo pyspark, por su parte, ofrece una interfaz para programar en Spark utilizando el lenguaje Python. La combinación de ambos permite realizar pruebas unitarias en cadenas de procesamiento en Spark.
Me gustaría escribir pruebas unitarias para métodos simples que contienen código de pyspark.
¿Cómo puedo simular la parte de lectura de spark? He intentado esto:
“`
@patch("class<em>to</em>test.SparkSession")
def test<em>do</em>stuff(self, mock<em>spark: MagicMock) -> None:
spark = MagicMock()
spark.read.return</em>value.format.return<em>value.load.return</em>value = \
self.spark.createDataFrame([(1, 2)], ["key2", "c2"])
mock<em>spark.return</em>value = spark
<pre><code>input_df = self.spark.createDataFrame([(1, 1)], ["key1", "c1"])
actual_df = ClassToTest().do_stuff(input_df,"df2","key1","key2")
expected_df = self.spark.createDataFrame([(1, 1, 1, 2)], ["key1", "c1", "key2", "c2"])
assert_pyspark_df_equal(actual_df, expected_df)
</code></pre>
<code>
Pero falla con este error:
</code>py4j.Py4JException: El método join([class java.util.ArrayList, class org.apache.spark.sql.Column, class java.lang.String]) no existe“`
Parece que la simulación no funcionó como esperaba, ¿qué debo hacer para que spark.read.load devuelva el dataframe de prueba que especifiqué?
Editar: código completo aquí
davy.ai
Para simular la parte de lectura de Spark, puedes utilizar el decorador
patch
del módulounittest.mock
para simular el métodospark.read
. Luego puedes encadenar los métodos simulados para devolver elDataFrame
que deseas probar.Aquí tienes un ejemplo de cómo puedes simular el método
spark.read
en tu prueba:En este ejemplo, estamos creando un DataFrame simulado (
mock_df2
) y configurándolo para devolver el DataFrame esperado cuando se llame al métodojoin
. Luego, estamos simulando el métodoread
del objetoSparkSession
para devolver el DataFrame simulado cuando se llame conload
. Finalmente, estamos creando nuestro DataFrame de entrada y llamando al método que se está probando (do_stuff
). Por último, estamos asegurando que los DataFrames esperados y reales sean iguales.