為Spark 程序添加單元測試

• 一、ScalaTest 測試框架
• 二、Spark Application 該如何進行單元測試？
• 三、Spark Fast Tests 組件

相比于傳統代碼，Spark是比較難調試的。程序運行在集群中，每次修改代碼后，都要上傳到集群進行測試，代價非常大，所以優先在本地進行單元測試，可以減少小模塊的邏輯錯誤。

一、ScalaTest 測試框架

ScalaTest是比JUnit和TestNG更加高階的測試編寫工具，這個Scala應用在JVM上運行，可以測試Scala以及Java代碼。ScalaTest一共提供了七種測試風格，分別為：FunSuite，FlatSpec，FunSpec，WordSpec，FreeSpec，PropSpec和FeatureSpec。
FunSuite的方式較為靈活，而且更符合傳統測試方法的風格，區別僅在于test()方法可以接受一個閉包。而FlatSpec和FunSpec則通過提供諸如it、should、describe等方法，來規定書寫測試的一種模式。
使用方式如下：
1、引入依賴

testCompile group: 'org.scalatest', name: 'scalatest_2.11', version: '3.0.6-SNAP5'

2、編寫對應的case

scala test 測試代碼

import org.scalatest._ class HelloWorldTestextends FunSuite {test("Test difference") {val a = Set("a", "b", "a", "c")val b = Set("b", "d")assert(a -- b === Set("a", "c"))}//交集test("Test intersection") {val a = Set("a", "b", "a", "c")val b = Set("b", "d")assert(a.intersect(b) === Set("b"))}//并集test("Test union") {val a = Set("a", "b", "a", "c")val b = Set("b", "d")assert(a ++ b === Set("a", "b", "c", "d"))}test("Test difference failed") {val a = Set("a", "b", "a", "c")val b = Set("c", "d")//應該等于Set("a","b")assert(a -- b === Set("a", "c"))} }

3、運行該test cast，查看執行效果

可以看到 前三條case 執行通過，最后一條未通過的case 也明確指出了期望值和實際值，這時候我們 根據提示去修改對應的處理邏輯代碼即可。

二、Spark Application 該如何進行單元測試？

通常一個完整的spark application，可以簡化的理解為是對一些數據的處理/運算 然后將運算的結果輸出到其它介質中。

因此在spark 程序中，會涉及到：
1、對數據的讀入（通過streming 或者 批量讀文件的形式）；
2、對數據進行運算（格式轉換、邏輯運算，篩選filter etc）；
3、對運算結果的輸出

在spark 2.x 的架構中，spark 程序入庫已經統一到SparkSession 、Dataset/DataFrame 為主要的用戶 API。我們所進行的數據計算操作都可以抽象的理解為是在對 Dataset 進行轉換操作，通過我們編寫的函數，從一個原 Dataset/DataFrame 得到一個我們所期望的目的 Dataset/DataFrame。

我們在編寫spark 程序的時候，應該盡量按照這個原則來構建spark 程序內部的功能模塊，即：輸入輸出獨立于業務計算，每個業務計算 通過Dataset/DataFrame 的轉換來封裝成不同的函數模塊。那么我們在進行單元測試的時候，就可以只重點關注到 我們實現的一個個具體業務計算的Dataset/DataFrame的轉換中，減少 輸入輸出 等環境依賴（輸入輸出的格式，依靠接口文檔）。

前面 提到的 ScalaTest 測試框架 更側重于對scala 語法本身的一些 集合，功能進行單元測試。上面我們分析了，對spark 進行單元測試的重心其實就是對 Dataset/DataFrame 轉換的測試。

三、Spark Fast Tests 組件

spark-fast-tests 是一個開源工具包，能夠很方便幫助我們 對比Dataset/DataFrame。其提供了以下一些方法：
兩個Dataset/DataFrame 相等對比
assertSmallDataFrameEquality 適用于本地較小兩個DF對比
assertSmallDataFrameEquality 適用于本地較小兩個DS對比
assertLargeDatasetEquality 適用于集群較大兩個DF對比
assertLargeDatasetEquality 適用于集群較大兩個DS對比
??相等比較時候 忽略 null 使用 ignoreNullable 參數
eg:
assertSmallDatasetEquality(actualDF, expectedDF，ignoreNullable=true)

一個 Dataset/DataFrame 兩列相等對比
assertColumnEquality(df, "Column1", "Column2")

使用實例：

import org.apache.spark.sql.SparkSession trait SparkSessionTestWrapper {lazy val spark: SparkSession = {SparkSession.builder().master("local").appName("spark test example").getOrCreate()} } import com.github.mrpowers.spark.fast.tests._ import org.scalatest.FunSpec class DatasetSpec extends FunSpec with SparkSessionTestWrapper with DatasetComparer with ColumnComparer {import spark.implicits._it("aliases a DataFrame") {val sourceDF = Seq(("jose"), ("li"),("luisa")).toDF("name")val actualDF = sourceDF.selectExpr("name as student")val expectedDF = Seq(("jose"),("li"),("luisa")).toDF("student")assertSmallDatasetEquality(actualDF, expectedDF)}it("df expected_name") {val df = Seq(("jose", "jose"),("li", "none"),("luisa", "luisa")).toDF("name", "expected_name")assertColumnEquality(df, "Column1", "Column2")} }

運行結果上，可以看到 第一條case 執行通過，最后一條未通過的case 也明確指出了期望值和實際值，這時候我們 根據提示去修改對應的處理邏輯代碼即可。
參考資料：
https://blog.csdn.net/hany3000/article/details/51033610
https://medium.com/@mrpowers/testing-spark-applications-8c590d3215fa
https://github.com/MrPowers/spark-fast-tests

總結

以上是生活随笔為你收集整理的为Spark 程序添加单元测试的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。