我的 Spark ETL 流程中有许多步骤，并且我创建了套件来测试每个步骤的功能，并且我的步骤的输出是下一步的输入。
但是，我想创建一个通过 ETL 过程运行两次的测试，例如：
原始数据 >> ETL_Step_1 >> ETL_Step_2 >> ETL_Step_3 >> 增量数据 >> ETL_Step_1 >> ETL_Step_2 >> ETL_Step_3
我从 Reporter 收到错误消息，因为它无法运行与以前运行过的套件同名的套件（例如，我无法两次运行名为 ETL_Step_1 的套件）。如果复制套件并重命名重复项以执行以下操作，问题就会消失：
原始数据 >> ETL_Step_1 >> ETL_Step_2 >> ETL_Step_3 >> 增量数据 >> ETL_Step_1_2 >> ETL_Step_2_2 >> ETL_Step_3_2
有没有更好的方法来运行我的测试而无需复制和重命名测试套件？
【问题讨论】：
您可以将所有 ETL 步骤放入定义（集合）并运行它们
import org.scalatest.FunSpec
import com.github.mrpowers.spark.fast.tests.DataFrameComparer
import com.github.mrpowers.spark.daria.sql.SparkSessionExt._
import org.apache.spark.sql.types.{IntegerType, StringType}
describe("etl collection") {
it("can run etls that are organized in a map") {
val sourceDF = spark.createDF(
List(
("bob", 14),
("liz", 20)
), List(
("name", StringType, true),
("age", IntegerType, true)
)
)
val etlDefinition = new EtlDefinition(
name = "example",
sourceDF = sourceDF, // etl step 1
transform = someTransform(), // etl step 2
write = someWriter(), // etl step 3
hidden = false
)
val etls = scala.collection.mutable.Map[String, EtlDefinition]("example" -> etlDefinition)
etls += ("ex2" -> etlDefinition)
etls("example").process()
}
}
另一个选项可能是来自 scalatest 的 eventually。最终的一个小例子：
Post("/something", body) ~> routes ~> check {
response should be a pendingResponse
eventually {
Post("/something", body) ~> routes ~> check {
response should be a expectedResponse
}
}
}
你可以调用一些方法来代替 Post。
更新：
it ("your definition") {
eventually { Thread.sleep(50); step1() should be A // assertion }
eventually { Thread.sleep(50); step2() should be B }
eventually { Thread.sleep(50); step3() should be C }
}
如果你愿意，你可以让它们嵌套
【讨论】：