apache-spark - Spark-shell 不允许查询结构化流

Question

我正在关注Spark the Definitive Guide这本书以下代码使用spark-shell在本地执行

程序:在没有任何其他选项的情况下启动 spark-shell

val static = spark.read.json("/part-00079-tid-730451297822678341-1dda7027-2071-4d73-a0e2-7fb6a91e1d1f-0-c000.json")
val dataSchema = static.schema
val streaming = spark.readStream.schema(dataSchema) .option("maxFilesPerTrigger",1).json("/part-00079-tid-730451297822678341-1dda7027-2071-4d73-a0e2-7fb6a91e1d1f-0-c000.json")
val activityCounts = streaming.groupBy("gt").count()
val activityQuery  = activityCounts.writeStream.queryName("activity_counts").format("memory").outputMode("complete").start()
activityQuery.awaitTermination()

书上是这么说的

After this code is executed the streaming computation will have started in the background

....

Now that this stream is running , we can experiment with the result by querying

我的观察:

执行此代码时，它不会释放 shell 让我输入命令比如

`spark.streams.active`

因此我不能查询这个流

我的研究

我尝试打开一个新的 spark-shell，但在该 shell 中查询没有返回任何结果。从这个 shell 获得的流是否可以从 shell 的其他另一个实例访问。

编辑1:

我想要内存中的表，以便我可以使用命令进行查询

for( i <- 1 to 5)
{
spark.sql("SELECT * FROM activity_counts").show()
Thread.sleep(1000)
}

Answer 1

几点:

1) 确保为Spark-shell分配了足够数量的Core

至少需要 2 个核心来运行流媒体/结构化流媒体应用程序以防止饥饿场景，即当流媒体应用程序启动时，1 个核心将分配给 Receiver 并且如果您启动只有 1 个核心的 spark 应用程序，将没有可用的核心来为执行者处理接收到的消息。

查看spark-shell中的核数:

spark.conf.get("spark.master")

启动4核spark-shell

spark-shell --master local[4]

2) 您正在将流写入内存，这不会在控制台中显示输出，要显示您必须注册表然后查询。

相反，您可以将格式从 memory 更改为 console 以在控制台中查看内容。

val activityQuery  = activityCounts.writeStream.queryName("activity_counts").format("console").outputMode("complete").start()