testing - 以编程方式在 apache spark 中创建 dstream

Question

我正在围绕 Apache Spark Streaming 编写一些独立的集成测试。我想测试我的代码是否可以在我的模拟测试数据中提取各种边缘情况。当我使用常规 RDD(不是流式传输)执行此操作时。我可以使用我的内联数据并对其调用“parallelize”以将其转换为 spark RDD。但是，我找不到这种创建流媒体的方法。理想情况下，我想偶尔调用一些“推送”函数，让 tupple 神奇地出现在我的 dstream 中。ATM 我通过使用 Apache Kafka 来做到这一点:我创建了一个临时队列，然后写入它。但这似乎有点矫枉过正。我宁愿直接从我的测试数据创建 test-dstream，而不必使用 Kafka 作为中介。

Answer 1

出于测试目的，您可以从 RDD 队列创建输入流。将更多 RDD 插入队列将模拟在批处理间隔内处理了更多事件。

val sc = SparkContextHolder.sc
val ssc = new StreamingContext(sc, Seconds(1))
val inputData: mutable.Queue[RDD[Int]] = mutable.Queue()
val inputStream: InputDStream[Int] = ssc.queueStream(inputData)

inputData += sc.makeRDD(List(1, 2)) // Emulate the RDD created during the first batch interval
inputData += sc.makeRDD(List(3, 4)) // 2nd batch interval
// etc

val result = inputStream.map(x => x*x)
result.foreachRDD(rdd => assertSomething(rdd))
ssc.start() // Don't forget to start the streaming context