Как написать модульные тесты в Spark 2.0+?

Ответ 1

Спасибо, что поставил этот выдающийся вопрос. По какой-то причине, когда дело доходит до Spark, каждый становится настолько увлечен аналитикой, что забывает о прекрасных методах разработки программного обеспечения, появившихся за последние 15 лет или около того. Вот почему мы предлагаем обсудить возможность тестирования и непрерывной интеграции (в том числе DevOps) в нашем курсе.

Быстрая поддержка терминологии

Истина unit test означает, что вы имеете полный контроль над каждым компонентом теста. Не может быть никакого взаимодействия с базами данных, вызовами REST, файловыми системами или даже системными часами; все должно быть "удвоено" (например, издеваться, прореживаться и т.д.), поскольку Джерард Мезарос помещает его в xUnit Test Patterns. Я знаю, что это похоже на семантику, но это действительно важно. Неспособность понять это - одна из основных причин, по которым вы видите прерывистые провалы при непрерывной интеграции.

Мы все еще можем Unit Test

Поэтому, учитывая это понимание, модульное тестирование RDD невозможно. Тем не менее, все еще существует место для модульного тестирования при разработке аналитики.

Рассмотрим простую операцию:

rdd.map(foo).map(bar)

Здесь foo и bar - простые функции. Те, кто может быть протестирован в обычном режиме, должны иметь столько угловых дел, сколько вы можете собрать. В конце концов, почему они заботятся о том, откуда они получают свои исходные данные, будь то тестовое устройство или RDD?

Не забудьте об искровой оболочке

Это не само по себе тестирование, но на этих ранних этапах вам также следует экспериментировать в оболочке Spark, чтобы выяснить ваши трансформации и особенно последствия вашего подхода. Например, вы можете изучить физические и логические планы запросов, стратегию разбиения и сохранение и состояние ваших данных со многими различными функциями, такими как toDebugString, explain, glom, show, printSchema и так далее. на. Я позволю вам изучить их.

Вы также можете установить ваш мастер в local[2] в оболочке Spark и в тестах для выявления любых проблем, которые могут возникнуть только после начала работы.

Тестирование интеграции с помощью Spark

Теперь для забавного материала.

Чтобы интегрировать тест Spark после уверенности в качестве ваших вспомогательных функций и логики преобразования RDD/DataFrame, важно сделать несколько вещей (независимо от инструмента сборки и тестовой среды):

• Увеличить JVM-память.
• Включить разблокировку, но отключить параллельное выполнение.
• Используйте тестовую среду для накопления тестов интеграции Spark в наборах и инициализируйте SparkContext перед всеми тестами и остановите ее после всех тестов.

С ScalaTest вы можете смешать в BeforeAndAfterAll (который я предпочитаю вообще) или BeforeAndAfterEach, как @ShankarKoirala делает для инициализации и срывания артефактов Spark. Я знаю, что это разумное место для исключения, но мне действительно не нравятся те mutable 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 в тест, а затем избавиться от него после его выполнения.

Страх-ориентированное программирование (спасибо, Натан)

Это все зависит от предпочтений, но я предпочитаю использовать шаблон займа и прокладывать себе дело до тех пор, пока я могу, прежде чем приносить другую структуру. Помимо того, что просто стараются оставаться в легком весе, фреймворки иногда добавляют много "магии", из-за которой трудно отлаживать отладочные тесты. Поэтому я подхожу к Suffering-Oriented Programming, где я избегаю добавления новой структуры до тех пор, пока боль от ее не будет слишком многого. Но опять же, это зависит от вас.

Лучший выбор для этой альтернативной структуры - это, конечно, spark-testing-base, как упоминалось в @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 -like и генераторы - если вы, конечно, понимаете, как работает тестирование на основе свойств и почему Это полезно. Но опять же, я лично не буду использовать его, пока моя аналитика и мои тесты не достигнут такого уровня сложности.

"Только ситы имеют дело с абсолютами". - Оби-Ван Кеноби

Конечно, вам не нужно выбирать тот или иной. Возможно, вы могли бы использовать подход Loan Pattern для большинства ваших тестов и искробезопасности только для нескольких, более строгих тестов. Выбор не двоичный; вы можете сделать и то и другое.

Тестирование интеграции с использованием Spark Streaming

Наконец, я просто хотел бы представить фрагмент того, что может быть установлено с помощью теста тестирования интеграции SparkStreaming со значениями в памяти без искробезопасности:

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. Большинство из них - это просто настройка шаблонов, которая работает с API-интерфейсом Spark. Независимо от того, вы можете сравнить это с StreamingSuiteBase как найдено в spark-testing-base, чтобы решить, что вы предпочитаете.

Это может быть мой самый длинный пост, поэтому я оставлю его здесь. Надеюсь, что другие услышат другие идеи, которые помогут улучшить качество нашей аналитики с помощью той же гибкой практики разработки программного обеспечения, которая улучшила все другие разработки приложений.

И извиняясь за бесстыдный плагин, вы можете проверить наш курс Analytics с Apache Spark, где мы рассматриваем много этих идей и Больше. Мы надеемся, что скоро будет онлайн-версия.

Ответ 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}

Холден Карау написал действительно хороший тест spark-testing-base

Вы должны проверить ниже, это простой пример

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

Надеюсь, это поможет!

Ответ 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.

Обновить

Библиотека искробезопасной базы автоматически добавляет SparkSession, когда некоторые признаки смешиваются с тестовым классом (например, когда DataFrameSuiteBase смешивается, у вас будет доступ к SparkSession через переменную spark).

Я создал отдельную тестовую библиотеку под названием 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)

    }

  }

}

Ответ 4

Начиная с Spark 1.6 вы можете использовать SharedSparkContext или SharedSQLContext который Spark использует для своих собственных модульных тестов:

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, где есть огромный набор различных тестовых наборов.

Ответ 5

Я мог бы решить проблему с помощью кода ниже

зависимость от искры-улья добавлена в проект pom

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

Ответ 6

Другой способ модульного тестирования с использованием 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)