默认（未指定）触发器如何确定结构化流中微批量的大小？

Question

当Spark Structured Streaming中的查询执行没有关于触发器的设置时，

import org.apache.spark.sql.streaming.Trigger

// Default trigger (runs micro-batch as soon as it can)
df.writeStream
  .format("console")
  //.trigger(???) // <--- Trigger intentionally omitted ----
  .start()

从 Spark 2.4.3（2019 年 8 月）开始。 Structured Streaming Programming Guide - Triggers 说

如果没有明确指定触发设置，那么默认情况下，查询将以微批处理模式执行，其中微批处理将在前一个微批处理完成后立即生成。

问题：默认触发器在什么基础上确定微批次的大小？

让我们说。输入源是 Kafka。由于一些中断，这项工作被中断了一天。然后重新启动相同的 Spark 作业。然后它将在它离开的地方消费消息。这是否意味着第一个微批次将是一个巨大的批次，其中包含 1 天的 msg 在作业停止时累积在 Kafka 主题中？假设这项工作需要 10 个小时来处理那个大批量，那么下一个微批量有 10 小时的消息？并逐渐直到 X 次迭代以赶上积压以到达更小的微批次。

Answer 1

默认触发器在什么基础上确定微批次的大小？

它没有。每个触发器（无论多长）都简单地请求输入数据集的所有源，并且它们提供的任何内容都由操作员在下游处理。消息来源知道要提供什么，因为他们知道到目前为止已经消耗（处理）了什么。

就好像您询问了批量结构化查询以及此单个“触发器”请求处理的数据大小（顺便说一句，有ProcessingTime.Once 触发器）。

这是否意味着第一个微批次将是一个巨大的批次，在作业停止时会在 Kafka 主题中累积 1 天的 msg？

几乎（与​​ Spark Structured Streaming 几乎没有任何关系）。

底层 Kafka 消费者要处理的记录数由 max.poll.records 配置，也可能由其他一些配置属性配置（参见 Increase the number of messages read by a Kafka consumer in a single poll）。

由于 Spark Structured Streaming 使用 Kafka 数据源，该数据源只是 Kafka Consumer API 的包装器，因此在单个微批处理中发生的任何事情都等同于单个 Consumer.poll 调用。

您可以使用带有 kafka. 前缀的选项（例如 kafka.bootstrap.servers）配置底层 Kafka 消费者，这些选项被考虑用于驱动程序和执行程序上的 Kafka 消费者。