Spark QueueStream 永不枯竭

Question

对我从互联网上借用的一段代码感到困惑，用于研究目的。这是代码：

import org.apache.spark.sql.SparkSession
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.{Seconds, StreamingContext}
import scala.collection.mutable

val spark = ... 

val sc = spark.sparkContext
val ssc = new StreamingContext(spark.sparkContext, Seconds(1)) 

val rddQueue = new mutable.Queue[RDD[Char]]()
val QS = ssc.queueStream(rddQueue) 

QS.foreachRDD(q=> {
   print("Hello") // Queue never exhausted
   if(!q.isEmpty) {
       ... do something
       ... do something
   }
}
)

//ssc.checkpoint("/chkpoint/dir") if unchecked causes Serialization error

ssc.start()
for (c <- 'a' to 'c') {
    rddQueue += ssc.sparkContext.parallelize(List(c))
}
ssc.awaitTermination()

我正在跟踪它只是为了检查并注意到 "hello" 正在永远打印出来：

 HelloHelloHelloHelloHelloHelloHelloHelloHelloHello and so on

我原以为 queueStream 会在 3 次迭代后耗尽。

那么，我错过了什么？

Answer 1

知道了。它实际上已经用完了，但循环仍在继续，这就是语句

的原因

 if(!q.isEmpty)

在那里。

好的，本来以为它会停止，或者不执行，但不是这样。我想起来了。根据批处理间隔的时间，如果没有流式传输，将产生一个空的 RDD。离开其他人，因为有人投票。

However, even though legacy, it is a bad example as adding checkpoint causes a Serialization error. Leaving it for the benefit of others.
ssc.checkpoint("/chkpoint/dir")

Spark QueueStream 永不枯竭

Spark QueueStream never exhausted

apache-spark

dstream