apache-spark - Structured Streaming 如何为每个微批处理规划流查询的逻辑计划？

Question

我在我的笔记本电脑上设置了一个小测试，它执行以下操作:

我创建了一个包含 1000 条消息的 Kafka 主题，其中每条消息包含几行，每行大约有 100 列。

在 List[Column] 中创建 300 个非常复杂的 Spark 列。没有聚合。

在从 Kafka 设置流时，我设置了 .option("maxOffsetsPerTrigger", 1) 所以在每个小批量中只处理一条消息。

然后我将这些列应用于仅包含一条消息的小批量。

val testStream = myStream
  .select(manyTestCols :_*)
  .writeStream
  .format("console")
  .outputMode("append")
  .start()
  .awaitTermination()

Spark 大约需要 10 秒来处理每条消息。

然后我将 maxOffsetsPerTrigger 更改为 .option("maxOffsetsPerTrigger", 1000) 以便在每个小批量中处理 1000 条消息。

Spark 需要大约 11 秒来处理每个小批量中的所有 1000 条消息。

因此，Spark 似乎做了某种“设置工作”，然后在每个小批量开始后非常快速地处理它。

对于每个小批量，这个“设置工作”是否会从查询计划到物理计划？

如果是这样，Spark 对每个小批量执行此操作是否有意义？

还是完全发生了其他事情？我正在查看 Spark 源代码，但希望得到已经完成此练习的人的反馈。

任何见解的 Tx。

Answer 1

Is this "setup work" going through the query planning through to the physical plan, for each mini-batch?

对于要在运行时填写的流查询的查询计划的执行特定部分部分是，如下(带有指向相应代码部分的链接):

Proper relations for data sources (例如 LocalRelation 用于无数据源)

Event-time watermark

Current (micro-batch) time

If so, does it make sense for Spark to do this each mini-batch?

绝对地。在 Structured Streaming 中没有其他方法可以短路无数据源，跟踪当前时间和水印。

这也是 extra no-data micro-batch for stateful operators 的原因例如，当水印发生变化时。

Am looking at Spark source code, but would appreciate feedback from someone that has gone through this exercise already.

见 MicroBatchExecution和 IncrementalExecution .