Wie schreibe ich Unit-Tests in Spark 2.0+?

Ich habe versucht, einen vernünftigen Weg zum Testen SparkSessionmit dem JUnit-Testframework zu finden. Obwohl es gute Beispiele dafür zu geben scheint SparkContext, konnte ich nicht herausfinden, wie ein entsprechendes Beispiel zum Laufen gebracht werden kann SparkSession, obwohl es an mehreren Stellen intern in der Funkenprüfbasis verwendet wird . Ich würde gerne eine Lösung ausprobieren, die auch keine Funkenprüfbasis verwendet, wenn dies nicht der richtige Weg ist.

Einfacher Testfall ( komplettes MWE-Projekt mit 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")
  }

}

Das Ergebnis dieser Ausführung mit JUnit ist eine NPE an der Lastlinie:

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)

Beachten Sie, dass es keine Rolle spielen sollte, ob die zu ladende Datei vorhanden ist oder nicht. In einer ordnungsgemäß konfigurierten SparkSession wird ein vernünftigerer Fehler ausgegeben .

Vielen Dank, dass Sie diese offene Frage gestellt haben. Wenn es um Spark geht, sind alle aus irgendeinem Grund so in die Analyse verwickelt, dass sie die großartigen Praktiken der Softwareentwicklung vergessen, die in den letzten 15 Jahren oder so entstanden sind. Aus diesem Grund legen wir Wert darauf, in unserem Kurs das Testen und die kontinuierliche Integration (unter anderem DevOps) zu diskutieren.

Ein kurzer Überblick über die Terminologie

Ein echter Komponententest bedeutet, dass Sie die vollständige Kontrolle über jede Komponente im Test haben. Es kann keine Interaktion mit Datenbanken, REST-Aufrufen, Dateisystemen oder sogar der Systemuhr geben. Alles muss "verdoppelt" werden (z. B. verspottet, gestoppt usw.), wie Gerard Mezaros es in xUnit-Testmustern formuliert . Ich weiß, das scheint Semantik zu sein, aber es ist wirklich wichtig. Wenn Sie dies nicht verstehen, ist dies ein Hauptgrund, warum bei der kontinuierlichen Integration zeitweise Testfehler auftreten.

Wir können noch Unit-Test

Angesichts dieses Verständnisses ist ein Unit-Test RDDunmöglich. Bei der Entwicklung von Analysen gibt es jedoch noch einen Platz für Unit-Tests.

Betrachten Sie eine einfache Operation:

rdd.map(foo).map(bar)

Hier foound barsind einfache Funktionen. Diese können auf normale Weise auf Einheit getestet werden, und sie sollten mit so vielen Eckkoffern wie möglich versehen sein. Warum interessiert es sie schließlich, woher sie ihre Eingaben beziehen, ob es sich um ein Testgerät oder ein Testgerät handelt RDD?

Vergessen Sie nicht die Spark Shell

Dies ist kein Test an sich , aber in diesen frühen Phasen sollten Sie auch in der Spark-Shell experimentieren, um Ihre Transformationen und insbesondere die Konsequenzen Ihres Ansatzes herauszufinden. Zum Beispiel können Sie physische und logische Abfragepläne, die Partitionierung der Strategie und die Erhaltung und den Zustand Ihrer Daten mit vielen verschiedenen Funktionen wie untersuchen toDebugString, explain, glom, show, printSchema, und so weiter. Ich werde Sie diese erkunden lassen.

Sie können Ihren Master auch local[2]in der Spark-Shell und in Ihren Tests festlegen , um Probleme zu identifizieren, die möglicherweise erst auftreten, wenn Sie mit der Verteilung der Arbeit beginnen.

Integrationstests mit Spark

Nun zu den lustigen Sachen.

Um Spark zu integrieren, nachdem Sie sich von der Qualität Ihrer Hilfsfunktionen und RDD/ oder der DataFrameTransformationslogik überzeugt haben , ist es wichtig, einige Dinge zu tun (unabhängig von Build-Tool und Test-Framework):

• Erhöhen Sie den JVM-Speicher.
• Aktivieren Sie das Gabeln, aber deaktivieren Sie die parallele Ausführung.
• Verwenden Sie Ihr Testframework, um Ihre Spark-Integrationstests in Suites zu akkumulieren, die SparkContextvor allen Tests zu initialisieren und nach allen Tests zu stoppen.

Mit ScalaTest können Sie mischen BeforeAndAfterAll(was ich allgemein bevorzuge) oder BeforeAndAfterEachwie @ShankarKoirala, um Spark-Artefakte zu initialisieren und abzubauen. Ich weiß, dass dies ein vernünftiger Ort ist, um eine Ausnahme zu machen, aber ich mag die veränderlichen vars, die Sie verwenden müssen, wirklich nicht .

Das Kreditmuster

Ein anderer Ansatz ist die Verwendung des Kreditmusters .

Zum Beispiel (mit 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()
  }
} 

Wie Sie sehen können, verwendet das Kreditmuster Funktionen höherer Ordnung, SparkContextum den Test zu "leihen" und ihn anschließend zu entsorgen.

Leidensorientierte Programmierung (Danke, Nathan)

Es ist völlig eine Frage der Präferenz, aber ich bevorzuge es, das Kreditmuster zu verwenden und die Dinge so lange wie möglich selbst zu verkabeln, bevor ich ein anderes Framework einbringe. Abgesehen davon, dass Frameworks nur versuchen, leicht zu bleiben, fügen sie manchmal viel "Magie" hinzu, was das Debuggen von Testfehlern schwierig macht. Daher verfolge ich einen leidensorientierten Programmieransatz, bei dem ich es vermeide, ein neues Framework hinzuzufügen, bis der Schmerz, es nicht zu haben, zu groß ist, um es zu ertragen. Aber auch dies liegt an Ihnen.

Die beste Wahl für dieses alternative Framework ist natürlich die Funkenprüfungsbasis, wie @ShankarKoirala erwähnt. In diesem Fall würde der obige Test folgendermaßen aussehen:

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
        }
      }
 }

Beachten Sie, dass ich nichts tun musste, um mit dem umzugehen SparkContext. SharedSparkContextgab mir das alles - mit scals SparkContext- kostenlos. Persönlich würde ich diese Abhängigkeit jedoch nicht nur für diesen Zweck einbringen, da das Kreditmuster genau das tut, was ich dafür brauche. Bei so viel Unvorhersehbarkeit, die bei verteilten Systemen auftritt, kann es ein echtes Problem sein, die Magie im Quellcode einer Drittanbieter-Bibliothek nachvollziehen zu müssen, wenn bei der kontinuierlichen Integration Probleme auftreten.

Jetzt, wo die Funkenprüfbasis wirklich glänzt, sind die Hadoop-basierten Helfer wie HDFSClusterLikeund YARNClusterLike. Das Einmischen dieser Eigenschaften kann Ihnen wirklich viel Setup-Schmerz ersparen. Ein weiterer Ort, an dem es glänzt, sind die Scalacheck- ähnlichen Eigenschaften und Generatoren - vorausgesetzt natürlich, Sie verstehen, wie eigenschaftsbasiertes Testen funktioniert und warum es nützlich ist. Aber auch hier würde ich mich persönlich zurückhalten, bis meine Analysen und Tests diesen Grad an Raffinesse erreicht haben.

"Nur ein Sith handelt absolut." -- Obi Wan Kenobi

Natürlich müssen Sie auch nicht das eine oder andere wählen. Vielleicht könnten Sie den Kreditmusteransatz für die meisten Ihrer Tests und die Funkenprüfungsbasis nur für einige strengere Tests verwenden. Die Wahl ist nicht binär; Sie können beides tun.

Integrationstests mit Spark Streaming

Abschließend möchte ich nur einen Ausschnitt davon präsentieren, wie ein SparkStreaming-Integrationstest-Setup mit speicherinternen Werten ohne Spark-Test-Basis aussehen könnte :

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

Das ist einfacher als es aussieht. Es verwandelt wirklich nur eine Folge von Daten in eine Warteschlange, um sie dem zuzuführen DStream. Das meiste davon ist wirklich nur ein Boilerplate-Setup, das mit den Spark-APIs funktioniert. Unabhängig davon können Sie dies mit dem StreamingSuiteBase in der Funkenprüfbasis gefundenen vergleichen , um zu entscheiden, welche Sie bevorzugen.

Dies könnte mein längster Beitrag sein, also werde ich ihn hier lassen. Ich hoffe, dass andere sich anderen Ideen anschließen, um die Qualität unserer Analysen mit denselben agilen Softwareentwicklungspraktiken zu verbessern, die alle anderen Anwendungsentwicklungen verbessert haben.

Und mit Entschuldigungen für den schamlosen Plug können Sie unseren Kurs Analytics with Apache Spark lesen , in dem wir viele dieser Ideen und mehr ansprechen. Wir hoffen, bald eine Online-Version zu haben.

Sie können einen einfachen Test mit FunSuite und BeforeAndAfterEach wie unten schreiben

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()
  }
}

Sie müssen im Test keine Funktionen erstellen, als die Sie einfach schreiben können

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

Holden Karau hat eine wirklich schöne Test- Funken-Test-Basis geschrieben

Sie müssen unten ein einfaches Beispiel auschecken

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)
  }
}

Hoffe das hilft!

Seit Spark 1.6 können Sie verwenden SharedSparkContextoder SharedSQLContextdass Spark für seine eigenen Unit-Tests verwendet:

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)
  }

Da Spark 2.3 SharedSparkSession verfügbar ist:

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)
  }

AKTUALISIEREN:

Maven-Abhängigkeit:

<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-Abhängigkeit:

"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"

Darüber hinaus können Sie die Testquellen von Spark überprüfen , bei denen es eine große Anzahl verschiedener Testanzüge gibt.

UPDATE 2:

Testen der Apache Spark-Einheit Teil 1 - Kernkomponenten

Apache Spark Unit Testing Teil 2 - Spark SQL

Apache Spark Unit Testing Teil 3 - Streaming

Testen der Apache Spark-Integration

Ich mag es, ein SparkSessionTestWrapperMerkmal zu erstellen, das in Testklassen eingemischt werden kann. Shankars Ansatz funktioniert, ist jedoch für Testsuiten mit mehreren Dateien unerschwinglich langsam.

import org.apache.spark.sql.SparkSession

trait SparkSessionTestWrapper {

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

}

Das Merkmal kann wie folgt verwendet werden:

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)

    }

  }

}

Überprüfen Sie das Spark-Spec- Projekt auf ein reales Beispiel, das den SparkSessionTestWrapperAnsatz verwendet.

Aktualisieren

Die Spark-Testing-Base-Bibliothek fügt die SparkSession automatisch hinzu, wenn bestimmte Merkmale in die Testklasse eingemischt werden (z. B. wenn DataFrameSuiteBaseeingemischt, haben Sie über die sparkVariable Zugriff auf die SparkSession ).

Ich habe eine separate Testbibliothek namens spark-fast-tests erstellt , um den Benutzern die volle Kontrolle über die SparkSession zu geben, wenn sie ihre Tests ausführen. Ich denke nicht, dass eine Testhilfebibliothek die SparkSession setzen sollte. Benutzer sollten in der Lage sein, ihre SparkSession nach Belieben zu starten und zu stoppen (ich möchte eine SparkSession erstellen und sie während des gesamten Testsuite-Laufs verwenden).

Hier ist ein Beispiel für die funkenschnelle Testmethode assertSmallDatasetEqualityin Aktion:

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)

    }

  }

}

Ich könnte das Problem mit dem folgenden Code lösen

Die Spark-Hive-Abhängigkeit wird im Projekt pom hinzugefügt

class DataFrameTest extends FunSuite with DataFrameSuiteBase{
        test("test dataframe"){
        val sparkSession=spark
        import sparkSession.implicits._
        var df=sparkSession.read.format("csv").load("path/to/csv")
        //rest of the operations.
        }
        }

Ein weiterer Weg zum Unit Test mit JUnit

import org.apache.spark.sql.SparkSession
import org.junit.Assert._
import org.junit.{After, Before, _}

@Test
class SessionSparkTest {
  var spark: SparkSession = _

  @Before
  def beforeFunction(): Unit = {
    //spark = SessionSpark.getSparkSession()
    spark = SparkSession.builder().appName("App Name").master("local").getOrCreate()
    System.out.println("Before Function")
  }

  @After
  def afterFunction(): Unit = {
    spark.stop()
    System.out.println("After Function")
  }

  @Test
  def testRddCount() = {
    val rdd = spark.sparkContext.parallelize(List(1, 2, 3))
    val count = rdd.count()
    assertTrue(3 == count)
  }

  @Test
  def testDfNotEmpty() = {
    val sqlContext = spark.sqlContext
    import sqlContext.implicits._
    val numDf = spark.sparkContext.parallelize(List(1, 2, 3)).toDF("nums")
    assertFalse(numDf.head(1).isEmpty)
  }

  @Test
  def testDfEmpty() = {
    val sqlContext = spark.sqlContext
    import sqlContext.implicits._
    val emptyDf = spark.sqlContext.createDataset(spark.sparkContext.emptyRDD[Num])
    assertTrue(emptyDf.head(1).isEmpty)
  }
}

case class Num(id: Int)