如何为 Sparks new Structured Streaming 编写集成测试？答案

【问题标题】：How to write integration tests for Sparks new Structured Streaming?如何为 Sparks new Structured Streaming 编写集成测试？
【发布时间】：2018-09-06 08:38:55
【问题描述】：

尝试测试 Spark Structured Streams ...但失败了...如何正确测试它们？

我遵循了来自here 的一般 Spark 测试问题，我最接近的尝试是 [1] 看起来像：

import simpleSparkTest.SparkSessionTestWrapper
import org.scalatest.FunSpec  
import org.apache.spark.sql.types.{StringType, IntegerType, DoubleType, StructType, DateType}
import org.apache.spark.sql.streaming.OutputMode

class StructuredStreamingSpec extends FunSpec with SparkSessionTestWrapper {

  describe("Structured Streaming") {

    it("Read file from system") {

      val schema = new StructType()
        .add("station_id", IntegerType)
        .add("name", StringType)
        .add("lat", DoubleType)
        .add("long", DoubleType)
        .add("dockcount", IntegerType)
        .add("landmark", StringType)
        .add("installation", DateType)

      val sourceDF = spark.readStream
        .option("header", "true")
        .schema(schema)
        .csv("/Spark-The-Definitive-Guide/data/bike-data/201508_station_data.csv")
        .coalesce(1)

      val countSource = sourceDF.count()

      val query = sourceDF.writeStream
        .format("memory")
        .queryName("Output")
        .outputMode(OutputMode.Append())
        .start()
        .processAllAvailable()

      assert(countSource === 70)
    }

  }

}

遗憾的是，org.apache.spark.sql.AnalysisException: Queries with streaming sources must be executed with writeStream.start() 总是失败

我还在 spark-testing-base 存储库中找到了这个 issue，我想知道是否可以测试 Spark 结构化流？

我想要进行集成测试，甚至可能在顶部使用 Kafka 来测试检查点或特定的损坏数据场景。有人可以帮帮我吗？

最后但并非最不重要的一点是，我认为版本可能也是一个限制 - 我目前针对 2.1.0 进行开发，因为 Azure HDInsight 部署选项我需要它。如果这是拖累，自托管是一种选择。

【问题讨论】：

尝试.trigger(Trigger.Once()) 和query.awaitTermination() 看看是否会得到不同的结果spark.apache.org/docs/latest/…

标签： apache-spark integration-testing scalatest

【解决方案1】：

你解决了吗？

在通过调用 start() 开始执行之前，您正在对流数据帧执行 count()。如果你想要一个计数，这样做怎么样？

  sourceDF.writeStream
    .format("memory")
    .queryName("Output")
    .outputMode(OutputMode.Append())
    .start()
    .processAllAvailable()

  val results: List[Row] = spark.sql("select * from Output").collectAsList()
  assert(results.size() === 70)

【讨论】：

遗憾的是，我还有一个旧版本。你试过最新的吗？

【解决方案2】：

您还可以使用 @holdenk 测试库中的 StructuredStreamingBase 特征： https://github.com/holdenk/spark-testing-base/blob/936c34b6d5530eb664e7a9f447ed640542398d7e/core/src/test/2.2/scala/com/holdenkarau/spark/testing/StructuredStreamingSampleTests.scala

这是一个如何使用它的例子：

class StructuredStreamingTests extends FunSuite with SharedSparkContext with StructuredStreamingBase {

override implicit def reuseContextIfPossible: Boolean = true

test("add 3") {
    import spark.implicits._
    val input = List(List(1), List(2, 3))
    val expected = List(4, 5, 6)
    def compute(input: Dataset[Int]): Dataset[Int] = {
        input.map(elem => elem + 3)
    }
    testSimpleStreamEndState(spark, input, expected, "append", compute)
}}

【讨论】：