使用 Spark Streaming 时限制 Kafka 批量大小

Question

是否可以限制 Kafka 消费者为 Spark Streaming 返回的批次大小？

我之所以问，是因为我得到的第一批记录有数亿条记录，并且需要很长时间来处理和检查它们。

Answer 1

限制最大批量大小有助于控制处理时间，但会增加消息的处理延迟。

通过属性下面的设置，我们可以控制批量大小 spark.streaming.receiver.maxRate= spark.streaming.kafka.maxRatePerPartition=

您甚至可以通过启用背压来根据处理时间动态设置批处理大小 spark.streaming.backpressure.enabled：真 spark.streaming.backpressure.initialRate：

Answer 2

除了上面的答案。批量大小是 3 个参数的乘积

1. batchDuration：流数据分批的时间间隔（以秒为单位）。
2. spark.streaming.kafka.maxRatePerPartition：设置每个分区每秒的最大消息数。这与batchDuration 结合使用将控制批量大小。您希望将maxRatePerPartition 设置为很大（否则您实际上是在限制您的工作），而batchDuration 则非常小。
3. kafka 主题中的分区数

为了更好地解释此产品在启用/禁用背压时如何工作 (set spark.streaming.kafka.maxRatePerPartition for createDirectStream)

Answer 3

我认为Spark Streaming Backpressure可以解决您的问题。

检查spark.streaming.backpressure.enabled 和spark.streaming.backpressure.initialRate。

默认情况下spark.streaming.backpressure.initialRate 未设置，spark.streaming.backpressure.enabled 默认情况下禁用，所以我想 spark 会尽可能多地使用。

来自Apache Spark Kafka configuration

spark.streaming.backpressure.enabled:

这使 Spark Streaming 能够根据接收速率控制 关于当前的批处理调度延迟和处理时间，以便 系统只能以系统可以处理的速度接收。 在内部，这会动态设置最大接收速率 接收器。这个比率的上限是值 spark.streaming.receiver.maxRate 和 spark.streaming.kafka.maxRatePerPartition 如果已设置（见下文）。

而且由于您想控制第一批，或者更具体地-第一批的消息数量，我认为您需要spark.streaming.backpressure.initialRate

spark.streaming.backpressure.initialRate:

这是每个接收器的初始最大接收速率 在背压机制启动时接收第一批数据 已启用。

当您的 Spark 作业（分别是 Spark 工作人员）能够处理来自 kafka 的 10000 条消息，但 kafka 代理向您的作业提供 100000 条消息时，这个很好。

也许您也有兴趣查看spark.streaming.kafka.maxRatePerPartition 以及Jeroen van Wilgenburg on his blog 对这些属性的真实示例的一些研究和建议。