控制台接收器不是基于笔记本的交互式工作流的好选择。即使在可以捕获输出的 Scala 中，它也需要在同一段落中调用 awaitTermination（或等效项），从而有效地阻止注释。
%spark
spark
.readStream
.format("socket")
.option("host", "localhost")
.option("port", "9999")
.option("includeTimestamp", "true")
.load()
.writeStream
.outputMode("append")
.format("console")
.option("truncate", "false")
.start()
.awaitTermination() // Block execution, to force Zeppelin to capture the output
链式awaitTermination 可以替换为独立调用在同一段落中也可以：
%spark
val query = df
.writeStream
...
.start()
query.awaitTermination()
没有它，Zeppelin 没有理由等待任何输出。 PySpark 只是在此之上增加了另一个问题 - 间接执行。因此，即使在这里阻止查询也无济于事。
此外，在浏览笔记时，流的连续输出可能会导致呈现问题和内存问题（可以通过InterpreterContext 或 REST API 使用 Zeppelin 显示系统，以实现更明智的行为，其中输出为被覆盖或定期清除）。
使用 Zeppelin 进行测试的更好选择是 memory sink。这样您就可以在不阻塞的情况下启动查询：
%pyspark
query = (windowedCounts
.writeStream
.outputMode("complete")
.format("memory")
.queryName("some_name")
.start())
并在另一个段落中按需查询结果：
%pyspark
spark.table("some_name").show()
它可以与reactive streams 或类似的解决方案结合使用，以提供基于间隔的更新。
也可以使用 StreamingQueryListener 和 Py4j 回调来耦合 rx 和 onQueryProgress 事件，尽管 PySpark 不支持查询侦听器，并且需要一些代码来将事情粘合在一起。 Scala 接口：
package com.example.spark.observer
import org.apache.spark.sql.streaming.StreamingQueryListener
import org.apache.spark.sql.streaming.StreamingQueryListener._
trait PythonObserver {
def on_next(o: Object): Unit
}
class PythonStreamingQueryListener(observer: PythonObserver)
extends StreamingQueryListener {
override def onQueryProgress(event: QueryProgressEvent): Unit = {
observer.on_next(event)
}
override def onQueryStarted(event: QueryStartedEvent): Unit = {}
override def onQueryTerminated(event: QueryTerminatedEvent): Unit = {}
}
构建一个 jar，调整构建定义以反映所需的 Scala 和 Spark 版本：
scalaVersion := "2.11.8"
val sparkVersion = "2.2.0"
libraryDependencies ++= Seq(
"org.apache.spark" %% "spark-sql" % sparkVersion,
"org.apache.spark" %% "spark-streaming" % sparkVersion
)
把它放在 Spark 类路径上，补丁 StreamingQueryManager:
%pyspark
from pyspark.sql.streaming import StreamingQueryManager
from pyspark import SparkContext
def addListener(self, listener):
jvm = SparkContext._active_spark_context._jvm
jlistener = jvm.com.example.spark.observer.PythonStreamingQueryListener(
listener
)
self._jsqm.addListener(jlistener)
return jlistener
StreamingQueryManager.addListener = addListener
启动回调服务器：
%pyspark
sc._gateway.start_callback_server()
并添加监听器：
%pyspark
from rx.subjects import Subject
class StreamingObserver(Subject):
class Java:
implements = ["com.example.spark.observer.PythonObserver"]
observer = StreamingObserver()
spark.streams.addListener(observer)
最后你可以使用subscribe 并阻止执行：
%pyspark
(observer
.map(lambda p: p.progress().name())
# .filter() can be used to print only for a specific query
.subscribe(lambda n: spark.table(n).show() if n else None))
input() # Block execution to capture the output
最后一步应该在您开始流式查询后执行。
也可以跳过rx 并像这样使用最少的观察者：
class StreamingObserver(object):
class Java:
implements = ["com.example.spark.observer.PythonObserver"]
def on_next(self, value):
try:
name = value.progress().name()
if name:
spark.table(name).show()
except: pass
它比Subject 提供的控制要少一些（需要注意的是，这会干扰其他代码打印到标准输出，并且只能由removing listener 停止。使用Subject，您可以轻松地dispose @987654347 @观察者，一旦你完成），但在其他方面应该或多或少相同。
请注意，任何阻塞操作都足以捕获侦听器的输出，而不必在同一个单元格中执行。例如
%pyspark
observer = StreamingObserver()
spark.streams.addListener(observer)
和
%pyspark
import time
time.sleep(42)
将以类似的方式工作，在定义的时间间隔内打印表格。
为了完整起见，您可以实现StreamingQueryManager.removeListener。