用于 azure 数据湖安装路径的 spark 读/写功能的单元测试答案

【问题标题】：Unit test for spark read/write function to azure data lake mount path用于 azure 数据湖安装路径的 spark 读/写功能的单元测试
【发布时间】：2022-10-01 02:46:22
【问题描述】：

我对单元测试很陌生，我需要为火花读/写功能编写单元测试。

我有如下功能，例如：

def read_file(path, format, logger):
   df = spark.read.load(path, format = format)
   etc.....
   return df

注意：路径已安装。

现在我需要为此功能编写单元测试/模拟adls位置.

我已经检查了下面的链接，该链接使用补丁来模拟 blobServiceClient 但我不确定这是否对我有用，因为它使用了 python 的开放函数。

Trying to use patch to mock BlobServiceClient but the real class is called instead

甚至可以模拟adls，如果可以的话怎么办？

任何人都可以在这方面帮助我。任何帮助，将不胜感激。

您能否分享您的代码的最小可重现示例？我们到底应该模拟什么？ adls 是什么？
adls 是天蓝色的数据湖存储，我想模拟它，因为我只从 alds 读取文件并且它已安装。
您的单元测试目的是测试对 ADLS 的访问还是测试功能。你能提供一个本地路径吗？
经过大量搜索后，我发现我可以使用 @mock.patch(BlobServiceClient) 测试对 ADLS 的访问，但现在我想测试功能，并且要求该功能应该是自包含的，它不应该做任何连接，这就是为什么我需要模拟路径，无论它是本地的还是安装的。
@ArturoMartinez 如果您有任何想法，那将非常有帮助。

标签： python apache-spark pyspark pytest azure-storage-account

【解决方案1】：

确保 HDFS 在本地可用。如果您在 Windows 上运行测试，请确保阅读此自述文件https://github.com/steveloughran/winutils

此 repo 引用了此其他 repo 以获得较新版本。这 2 个 github 贡献者是 Hadoop 团队的成员，这些是我用于本地环境的二进制文件。

完成此设置后，您可以传递本地路径。它们可以是本地 linux 或 windows 路径。

另请阅读本文以正确设置 hadoop 二进制文件的 PATH 项：https://sparkbyexamples.com/spark/spark-hadoop-exception-in-thread-main-java-lang-unsatisfiedlinkerror-org-apache-hadoop-io-nativeio-nativeiowindows-access0ljava-lang-stringiz/?msclkid=4333eb3bc05d11ecb99904c712581198

【讨论】：

【解决方案2】：

Write UT for all DataFrameWriter, DataFrameReader, DataStreamReader, DataStreamWriter

使用上述步骤的示例测试用例

模拟
行为

断言

基于 Maven 的依赖

<groupId>org.scalatestplus</groupId>
<artifactId>mockito-3-4_2.11</artifactId>
<version>3.2.3.0</version>
<scope>test</scope>


<groupId>org.mockito</groupId>
<artifactId>mockito-inline</artifactId>
<version>2.13.0</version>
<scope>test</scope>

让我们使用一个 spark 类的示例，其中 source 是 Hive，sink 是 JDBC

class DummySource extends SparkPipeline {
  /**
   * Method to read the source and create a Dataframe
   *
   * @param sparkSession : SparkSession
   * @return : DataFrame
   */
  override def read(spark: SparkSession): DataFrame = {
    spark.read.table("Table_Name").filter("_2 > 1")
  }

  /**
   * Method to transform the dataframe
   *
   * @param df : DataFrame
   * @return : DataFrame
   */
  override def transform(df: DataFrame): DataFrame = ???

  /**
   * Method to write/save the Dataframe to a target
   *
   * @param df : DataFrame
   *
   */
  override def write(df: DataFrame): Unit =
    df.write.jdbc("url", "targetTableName", new Properties())
}

模拟阅读

test("Spark read table") {
  val dummySource = new DummySource()
  val sparkSession = SparkSession
    .builder()
    .master("local[*]")
    .appName("mocking spark test")
    .getOrCreate()
  val testData = Seq(("one", 1), ("two", 2))
  val df = sparkSession.createDataFrame(testData)
  df.show()
  val mockDataFrameReader = mock[DataFrameReader]
  val mockSpark = mock[SparkSession]
  when(mockSpark.read).thenReturn(mockDataFrameReader)
  when(mockDataFrameReader.table("Table_Name")).thenReturn(df)
  dummySource.read(mockSpark).count() should be(1)
}

模拟写

  test("Spark write") {
  val dummySource = new DummySource()
  val mockDf = mock[DataFrame]
  val mockDataFrameWriter = mock[DataFrameWriter[Row]]
  when(mockDf.write).thenReturn(mockDataFrameWriter)
  when(mockDataFrameWriter.mode(SaveMode.Append)).thenReturn(mockDataFrameWriter)
  doNothing().when(mockDataFrameWriter).jdbc("url", "targetTableName", new Properties())
  dummySource.write(df = mockDf)
}

参考中的流式代码

参考：https://medium.com/walmartglobaltech/spark-mocking-read-readstream-write-and-writestream-b6fe70761242

【讨论】：