如何在Spark 2.0+中编写单元测试？

Question

我一直试图找到一种合理的方法来测试SparkSessionJUnit测试框架.虽然似乎有很好的例子SparkContext,但我无法弄清楚如何使用相应的示例SparkSession,即使它在spark-testing-base内部的几个地方使用过.我很乐意尝试一种不使用spark-testing-base的解决方案,如果它不是真正正确的方式去这里.

简单的测试用例(完整MWE项目有build.sbt):

import com.holdenkarau.spark.testing.DataFrameSuiteBase
import org.junit.Test
import org.scalatest.FunSuite

import org.apache.spark.sql.SparkSession


class SessionTest extends FunSuite with DataFrameSuiteBase {

  implicit val sparkImpl: SparkSession = spark

  @Test
  def simpleLookupTest {

    val homeDir = System.getProperty("user.home")
    val training = spark.read.format("libsvm")
      .load(s"$homeDir\\Documents\\GitHub\\sample_linear_regression_data.txt")
    println("completed simple lookup test")
  }

}

使用JUnit运行它的结果是加载线上的NPE:

java.lang.NullPointerException
    at SessionTest.simpleLookupTest(SessionTest.scala:16)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
    at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
    at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
    at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
    at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)
    at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:78)
    at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:57)
    at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)
    at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)
    at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
    at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)
    at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)
    at org.junit.runners.ParentRunner.run(ParentRunner.java:363)
    at org.junit.runner.JUnitCore.run(JUnitCore.java:137)
    at com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:68)
    at com.intellij.rt.execution.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(IdeaTestRunner.java:51)
    at com.intellij.rt.execution.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:237)
    at com.intellij.rt.execution.junit.JUnitStarter.main(JUnitStarter.java:70)

请注意,加载的文件是否存在无关紧要; 在正确配置的SparkSession中,将抛出更明智的错误.

Answer 1

感谢您提出这个悬而未决的问题.出于某种原因,当谈到Spark时,每个人都会陷入分析之中,以至于忘记了过去15年左右出现的优秀软件工程实践.这就是我们在课程中讨论测试和持续集成(以及DevOps等)的原因.

快速学习术语

一个真正的单元测试意味着你有过在测试每个组件的完全控制.不能与数据库,REST调用,文件系统甚至系统时钟交互; Gerard Mezaros将其置于xUnit测试模式中,一切都必须"加倍"(例如模拟,存根等).我知道这看起来像语义,但它确实很重要.不理解这是您在持续集成中看到间歇性测试失败的一个主要原因.

我们还可以进行单元测试

因此,鉴于这种理解,单元测试RDD是不可能的.但是,在开发分析时仍有一个单元测试的地方.

考虑一个简单的操作:

rdd.map(foo).map(bar)

这里foo和bar简单的功能.这些可以通过正常的方式进行单元测试,并且它们应该与尽可能多的角落情况一样.毕竟,他们为什么要关心从输入测试装置到测试装置的位置RDD？

不要忘记Spark Shell

这本身并不是测试,但在这些早期阶段,您还应该在Spark shell中进行实验,以确定您的转换,尤其是您的方法的后果.例如,您可以检查物理和逻辑查询计划,分区策略和保存,以及您的数据中包含许多不同的功能状态toDebugString,explain,glom,show,printSchema,等.我会让你探索那些.

您还可以local[2]在Spark shell和测试中将master设置为可以识别在开始分发工作后可能出现的任何问题.

使用Spark进行集成测试

现在为有趣的东西.

为了在您对辅助函数和/ 转换逻辑的质量有信心之后集成测试 Spark ,做一些事情(不管构建工具和测试框架)是至关重要的:RDDDataFrame

• 增加JVM内存.
• 启用分叉但禁用并行执行.
• 使用您的测试框架将Spark集成测试累积到套件中,并SparkContext在所有测试之前初始化并在所有测试之后停止它.

使用ScalaTest,您可以混合BeforeAndAfterAll(我通常更喜欢)或BeforeAndAfterEach@ShankarKoirala来初始化和拆除Spark工件.我知道这是一个合理的例外地方,但我真的不喜欢var你必须使用的那些可变的.

贷款模式

另一种方法是使用贷款模式.

例如(使用ScalaTest):

class MySpec extends WordSpec with Matchers with SparkContextSetup {
  "My analytics" should {
    "calculate the right thing" in withSparkContext { (sparkContext) =>
      val data = Seq(...)
      val rdd = sparkContext.parallelize(data)
      val total = rdd.map(...).filter(...).map(...).reduce(_ + _)

      total shouldBe 1000
    }
  }
}

trait SparkContextSetup {
  def withSparkContext(testMethod: (SparkContext) => Any) {
    val conf = new SparkConf()
      .setMaster("local")
      .setAppName("Spark test")
    val sparkContext = new SparkContext(conf)
    try {
      testMethod(sparkContext)
    }
    finally sparkContext.stop()
  }
} 

正如您所看到的,贷款模式利用高阶函数"贷款" SparkContext到测试,然后在完成后处理它.

以苦难为导向的编程(谢谢,内森)

这完全是一个偏好的问题,但我更喜欢使用贷款模式,并在引入另一个框架之前尽可能地连接自己.除了试图保持轻量级之外,框架有时会添加许多"魔力",使调试测试失败难以推理.因此,我采取了一种面向苦难的编程方法 - 我避免添加一个新的框架,直到没有它的痛苦太多无法忍受.但同样,这取决于你.

正如@ShankarKoirala所提到的那样,替代框架的最佳选择当然是火花测试基础.在这种情况下,上面的测试看起来像这样:

class MySpec extends WordSpec with Matchers with SharedSparkContext {
      "My analytics" should {
        "calculate the right thing" in { 
          val data = Seq(...)
          val rdd = sc.parallelize(data)
          val total = rdd.map(...).filter(...).map(...).reduce(_ + _)

          total shouldBe 1000
        }
      }
 }

注意我怎么没有做任何事情来处理SparkContext.SharedSparkContext给了我所有的 - sc作为SparkContext- 免费.就个人而言,我不会为了这个目的而引入这种依赖,因为贷款模式完全符合我的需要.此外,由于分布式系统发生了如此多的不可预测性,因此在持续集成中出现问题时,必须追踪第三方库源代码中发生的魔法可能会非常痛苦.

现在,火花测试基础真正发挥作用的是基于Hadoop的助手,比如HDFSClusterLike和YARNClusterLike.混合这些特性可以真正为您节省大量的设置痛苦.它闪耀的另一个地方是类似Scalacheck的属性和生成器 - 当然,您应该了解基于属性的测试是如何工作的以及为什么它有用.但同样,我会亲自推迟使用它,直到我的分析和我的测试达到这种复杂程度.

"只有一个西斯交易绝对." - 欧比旺克诺比

当然,您也不必选择其中一个.也许你可以在大多数测试和火花测试基础上使用贷款模式方法,仅用于一些更严格的测试.选择不是二元的; 你可以做到这两点.

使用Spark Streaming进行集成测试

最后,我想提供一个片段,说明SparkStreaming集成测试设置与内存中的值可能看起来没有spark-testing-base:

val sparkContext: SparkContext = ...
val data: Seq[(String, String)] = Seq(("a", "1"), ("b", "2"), ("c", "3"))
val rdd: RDD[(String, String)] = sparkContext.parallelize(data)
val strings: mutable.Queue[RDD[(String, String)]] = mutable.Queue.empty[RDD[(String, String)]]
val streamingContext = new StreamingContext(sparkContext, Seconds(1))
val dStream: InputDStream = streamingContext.queueStream(strings)
strings += rdd

这比它看起来更简单.它实际上只是将一系列数据转换为一个队列来提供给DStream.其中大多数只是与Spark API一起使用的样板设置.无论如何,你可以比较这StreamingSuiteBase 如发现 火花试验基地,以决定您喜欢哪一种.

这可能是我有史以来最长的职位,所以我会留在这里.我希望其他人能够提出其他想法,通过改进所有其他应用程序开发的敏捷软件工程实践来帮助提高我们的分析质量.

对于无耻插件的道歉,您可以查看我们的课程Analytics with Apache Spark,我们在其中解决了很多这些想法等等.我们希望尽快有一个在线版本.

Answer 2

您可以使用FunSuite和BeforeAndAfterEach编写一个简单的测试,如下所示

class Tests extends FunSuite with BeforeAndAfterEach {

  var sparkSession : SparkSession = _
  override def beforeEach() {
    sparkSession = SparkSession.builder().appName("udf testings")
      .master("local")
      .config("", "")
      .getOrCreate()
  }

  test("your test name here"){
    //your unit test assert here like below
    assert("True".toLowerCase == "true")
  }

  override def afterEach() {
    sparkSession.stop()
  }
}

您不需要在测试中创建函数,只需编写为

test ("test name") {//implementation and assert}

Holden Karau编写了非常好的测试火花测试基础

您需要查看下面的一个简单示例

class TestSharedSparkContext extends FunSuite with SharedSparkContext {

  val expectedResult = List(("a", 3),("b", 2),("c", 4))

  test("Word counts should be equal to expected") {
    verifyWordCount(Seq("c a a b a c b c c"))
  }

  def verifyWordCount(seq: Seq[String]): Unit = {
    assertResult(expectedResult)(new WordCount().transform(sc.makeRDD(seq)).collect().toList)
  }
}

希望这可以帮助!

Answer 3

我喜欢创建一个SparkSessionTestWrapper可以混合到测试类中的特征.Shankar的方法很有效,但对于包含多个文件的测试套件来说,它的速度非常慢.

import org.apache.spark.sql.SparkSession

trait SparkSessionTestWrapper {

  lazy val spark: SparkSession = {
    SparkSession.builder().master("local").appName("spark session").getOrCreate()
  }

}

特征可以如下使用:

class DatasetSpec extends FunSpec with SparkSessionTestWrapper {

  import spark.implicits._

  describe("#count") {

    it("returns a count of all the rows in a DataFrame") {

      val sourceDF = Seq(
        ("jets"),
        ("barcelona")
      ).toDF("team")

      assert(sourceDF.count === 2)

    }

  }

}

检查spark-spec项目,了解使用该SparkSessionTestWrapper方法的真实示例.

更新

当某些特征混合到测试类中时,spark-testing-base库会自动添加SparkSession(例如,当DataFrameSuiteBase混入时,您可以通过spark变量访问SparkSession ).

我创建了一个名为spark-fast-tests的独立测试库,以便用户在运行测试时完全控制SparkSession.我不认为测试助手库应该设置SparkSession.用户应该能够按照自己的意愿启动和停止SparkSession(我喜欢创建一个SparkSession并在整个测试套件运行中使用它).

以下是运行中的spark-fast-tests assertSmallDatasetEquality方法的示例:

import com.github.mrpowers.spark.fast.tests.DatasetComparer

class DatasetSpec extends FunSpec with SparkSessionTestWrapper with DatasetComparer {

  import spark.implicits._

    it("aliases a DataFrame") {

      val sourceDF = Seq(
        ("jose"),
        ("li"),
        ("luisa")
      ).toDF("name")

      val actualDF = sourceDF.select(col("name").alias("student"))

      val expectedDF = Seq(
        ("jose"),
        ("li"),
        ("luisa")
      ).toDF("student")

      assertSmallDatasetEquality(actualDF, expectedDF)

    }

  }

}

Answer 4

从Spark 1.6开始,您可以使用SharedSparkContext或SharedSQLContextSpark用于自己的单元测试:

class YourAppTest extends SharedSQLContext {

  var app: YourApp = _

  protected override def beforeAll(): Unit = {
    super.beforeAll()

    app = new YourApp
  }

  protected override def afterAll(): Unit = {
    super.afterAll()
  }

  test("Your test") {
    val df = sqlContext.read.json("examples/src/main/resources/people.json")

    app.run(df)
  }

自Spark 2.3 SharedSparkSession可用:

class YourAppTest extends SharedSparkSession {

  var app: YourApp = _

  protected override def beforeAll(): Unit = {
    super.beforeAll()

    app = new YourApp
  }

  protected override def afterAll(): Unit = {
    super.afterAll()
  }

  test("Your test") {
    df = spark.read.json("examples/src/main/resources/people.json")

    app.run(df)
  }

更新:

Maven依赖:

<dependency>
  <groupId>org.scalactic</groupId>
  <artifactId>scalactic</artifactId>
  <version>SCALATEST_VERSION</version>
</dependency>
<dependency>
  <groupId>org.scalatest</groupId>
  <artifactId>scalatest</artifactId>
  <version>SCALATEST_VERSION</version>
  <scope>test</scope>
</dependency>
<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-core</artifactId>
  <version>SPARK_VERSION</version>
  <type>test-jar</type>
  <scope>test</scope>
</dependency>
<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-sql</artifactId>
  <version>SPARK_VERSION</version>
  <type>test-jar</type>
  <scope>test</scope>
</dependency>

SBT依赖:

"org.scalactic" %% "scalactic" % SCALATEST_VERSION
"org.scalatest" %% "scalatest" % SCALATEST_VERSION % "test"
"org.apache.spark" %% "spark-core" % SPARK_VERSION % Test classifier "tests"
"org.apache.spark" %% "spark-sql" % SPARK_VERSION % Test classifier "tests"

此外,您可以检查Spark的测试源,其中有大量各种测试套装.