Test unitaire de la diffusion en continu

Question

J'ai cette ligne en scala :

class SparkStreamingService(...){
 val lines: org.apache.spark.streaming.dstream.InputDStream[(String, String)] = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
    ssc, kafkaParams, topics)
  lines.foreachRDD { rdd =>
    val df = cassandraSQLContext.read.json(rdd.map(x => x._2))
    if (!df.rdd.isEmpty()) {
      processDataFrameToLoadService(df)
    }
    else{
      throw new BacSparkStreamingExpception("The dataframe created from kafka message is empty")
    }
  }

  ssc.start()
  ssc.awaitTermination()
}

Existe-t-il un moyen de simuler un InputDStream en Java ? Comment puis-je procéder pour effectuer des tests unitaires ? En gros, je veux me moquer de cassandraSQLContext.read.json(rdd.map(x => x._2)) et retourner un DF personnalisé (ce que je peux faire), mais comment faire pour dépasser les 3 premières lignes ? Actuellement, il ne fait qu'interroger les messages et le test ne se termine jamais.

Answer 1

Un véritable test unitaire dans un contexte Spark est pratiquement impossible. Je pense que vous devriez plutôt test d'intégration avec des valeurs en mémoire :

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

...et partir de là.

Vous voulez également initialiser autant de choses Spark que possible dans une sorte de "configuration" ou "avant" fonctionnalité fournie par votre cadre de test - et de la même manière le démanteler dans un "démantèlement" ou "après".

Bien qu'ils ne soient pas strictement des tests unitaires, ces tests vous aideront à être sûr que votre code de traitement peut gérer les flux, et si les choses tournent mal lorsque la source des flux devient réelle ( c'est-à-dire Kafka), vous pouvez alors chercher la cause du problème du côté de Kafka.