apache-spark - 如何理解apache spark中的queueStream API？

Question

pyspark 有一个 api queueStream 用于从一系列 rdd 构造 dstream。

queueStream(rdds, oneAtATime=True, default=None)
Create an input stream from an queue of RDDs or list. In each batch, it will process either one or all of the RDDs returned by the queue.

NOTE: changes to the queue after the stream is created will not be recognized.

Parameters: 
rdds – Queue of RDDs
oneAtATime – pick one rdd each time or pick all of them once.
default – The default rdd if no more in rdds

问题 1:

在分布式环境中，如果我定义了一个队列对象 q1.我做q1.add(RDD)之类的操作。 q1 对象是否会转移到所有工作节点？ q1.add(RDD)这个对象复制到其他节点上会不会有问题？

问题2:

在我运行 dstream = queueStream(q1) 之后。
如果我继续将 RDD 放入队列中。这些 RDDS 会添加到 dstream 中吗？

Answer 1

我相信以下注意事项:

changes to the queue after the stream is created will not be recognized.

几乎可以回答这个问题，但要了解为什么会出现这种情况，您必须在 PySpark 代码中这样做，尤其是 following line :

queue = self._jvm.PythonDStream.toRDDQueue([r._jrdd for r in rdds])

如果这还不够，您可以查看 the corresponding Scala code 以查看它需要一个静态列表:

def toRDDQueue(rdds: JArrayList[JavaRDD[Array[Byte]]])

并将其转换为队列:

val queue = new java.util.LinkedList[JavaRDD[Array[Byte]]]
rdds.asScala.foreach(queue.add)

因此，Python 端的任何更改根本无法反射(reflect)在流中。

关于第一个问题，答案是否定的。队列不会被分发，因为 RDD 在 Driver 上下文之外根本没有意义。

注意 :

需要明确的是，Scala queueStream 将反射(reflect)添加到队列中的内容。 There is even an example in the Spark source