迭代 Stream 数据帧中的列值，并使用 Scala 和 Spark 将每个值分配给一个公共列表答案

【问题标题】：Iterate a column values in a Stream dataframe and assign each value to a common list using Scala and Spark迭代 Stream 数据帧中的列值，并使用 Scala 和 Spark 将每个值分配给一个公共列表
【发布时间】：2021-06-19 17:35:39
【问题描述】：

我有以下 Stream 数据框

+------------------------------------+
|______sentence______________________|
| Representative is a scientist      |
| Norman did a good job in the exam  |
| you want to go on shopping?        |
--------------------------------------

我有如下列表

val myList

作为最终输出，我需要 myList 在流数据帧中包含以上三个句子

输出

myList = [Representative is a scientist, Norman did a good job in the exam, you want to go on shopping? ]

我尝试了以下给出流错误的方法

val myList =   sentenceDataframe.select("sentence").rdd.map(r => r(0)).collect.toList

上述方法抛出的错误

org.apache.spark.sql.AnalysisException：带有流式源的查询必须用 writeStream.start() 执行

请注意，上述方法适用于普通数据帧，但不适用于流数据帧。

有没有办法遍历流数据帧的每一行并使用 scala 和 spark 将行值分配到公共列表中？

【问题讨论】：

标签： list scala dataframe apache-spark stream

【解决方案1】：

这听起来像是一个非常奇怪的用例，因为理论上流可能永远不会结束。你确定你不只是在寻找常见的 spark DataFrames 吗？

如果不是这种情况，您可以做的是使用累加器和火花流 foreachBatch 接收器。我使用了一个简单的套接字连接来演示这一点。您可以在例如下启动一个简单的套接字服务器。带有 nc -lp 3030 的 ubuntu 并且刚刚将消息传递到流中，生成的 DataFrame 将具有 [value: String] 的模式

val acc = spark.sparkContext.collectionAccumulator[String]

val stream = spark.readStream.format("socket").option("host", "localhost").option("port", "3030").load()

val query = stream.writeStream.foreachBatch((df: DataFrame, l: Long) => {
     df.collect.foreach(v => acc.add(v(0).asInstanceOf[String]))
  }).start()

...

// For some reason you are stopping the stream here
query.stop()
val myList = acc.value

现在你可能有一个问题是为什么我们使用累加器而不仅仅是一个 ArrayBuffer。 ArrayBuffers 可以在本地工作，但在集群上，foreachBatch 中的代码可能会在完全不同的节点上执行。这意味着它不会产生任何影响，这也是累加器首先存在的原因（参见https://spark.apache.org/docs/latest/rdd-programming-guide.html#accumulators）

【讨论】：