Packages

•  package root

  Definition Classes

  root

•  package cloudflow

  Definition Classes

  root

•  package spark

  Definition Classes

  cloudflow

•  package testkit

  Definition Classes

  spark

•  final case class SparkStreamletTestkit(session: SparkSession, config: Config = ConfigFactory.empty, maxDuration: Duration = 30.seconds) extends Product with Serializable

  Testkit for testing Spark streamlets.

  Testkit for testing Spark streamlets.

  The steps to write a test using the testkit are:

  1. Create the test class and extend it with SparkScalaTestSupport
  2. Create the Spark streamlet testkit instance
  3. Create the Spark streamlet
  4. Setup inlet tap on inlet port
  5. Setup outlet tap on outlet port
  6. Build input data and send to inlet tap
  7. Run the test
  8. Get data from outlet and assert

  // 1. Create the test class and extend it with `SparkScalaTestSupport`
  class MySparkStreamletSpec extends SparkScalaTestSupport {
  
   "SparkProcessor" should {
     "process streaming data" in {
  
       // 2. Create Spark streamlet testkit instance
       val testKit = SparkStreamletTestkit(session)
  
       // 3. Create spark streamlet
       val processor = new SparkProcessor[Data, Simple] {
         override def createLogic(): ProcessorLogic[Data, Simple] = new ProcessorLogic[Data, Simple](OutputMode.Append) {
           override def process(inDataset: Dataset[Data]): Dataset[Simple] =
             inDataset.select($"name").as[Simple]
         }
       }
  
       // 4. Setup inlet(s) tap on inlet port(s)
       val in: SparkInletTap[Data] = inletAsTap[Data](processor.shape.inlet)
  
       // 5. Setup outlet tap(s) on outlet port(s)
       val out: SparkOutletTap[Simple] = outletAsTap[Simple](processor.shape.outlet)
  
       // 6. Prepare input data and send it to the inlet tap(s)
       val data = (1 to 10).map(i => Data(i, s"name$i"))
       in.addData(data)
  
       // 7. Run the test
       run(processor, Seq(in), Seq(out))
  
       // 8. Get data from outlet tap(s) and assert
       val results = out.asCollection(session)
  
       results should contain(Simple("name1"))
     }
   }
  }

  Note: Every test is executed against a SparkSession which gets created and removed as part of the test lifecycle methods.

  Definition Classes

  testkit

• InletTapOps
• OutletTapOps

cloudflow.spark.testkit.SparkStreamletTestkit

OutletTapOps 