Spark Streaming对接Kafka时背压与maxRatePerPartition参数疑问

前置说明 #

你提到的spark.streaming.*前缀的参数均属于*Spark Streaming（DStream API，Spark 2.0之前的旧流处理API）*的配置，你当前代码使用的是Structured Streaming，对应参数体系有差异，以下回答先基于Spark Streaming的参数逻辑解答你的疑问，最后补充Structured Streaming的对应配置方案。

问题1：spark.streaming.receiver.maxRate与spark.streaming.kafka.maxRatePerPartition的关联 #

• 适用场景差异：
  • spark.streaming.receiver.maxRate：针对基于Receiver的Kafka消费模式，限制单个Receiver进程每秒最多接收的消息总数，是Receiver维度的全局上限
  • spark.streaming.kafka.maxRatePerPartition：针对无Receiver的Direct Kafka消费模式，限制单个Kafka分区每秒最多拉取的消息数，是分区维度的细粒度上限
• 关联逻辑：两个参数都是背压动态速率的上限阈值，背压算法根据历史批次指标计算出的推荐接收速率，不能超过两个参数限定的最大值，最终生效上限取两个参数限制的最小值。举个例子：假设1个Receiver对应3个Kafka分区，spark.streaming.receiver.maxRate设为1000条/秒，spark.streaming.kafka.maxRatePerPartition设为200条/秒/分区，那么单Receiver每秒最多拉取3*200=600条，小于1000的全局上限，最终生效的速率上限就是600条/秒。

问题2：首次运行无历史数据时是否需要配置spark.streaming.backpressure.initialRate？如何取值？ #

• 是否需要配置：分场景判断
  • 如果作业启动时Kafka对应分区有大量消息积压，必须配置，否则首次拉取没有速率限制，直接拉取远超处理能力的消息，会导致作业OOM、批次延迟飙升甚至崩溃
  • 如果作业从latest偏移开始消费，启动时无积压消息，可以不配置
• 取值方法：先压测得到作业单批次稳定处理的最大消息数，除以批次间隔得到每秒可处理的总消息数，再除以消费的分区总数，得到单分区每秒可处理的消息数，保守点打7-8折就是initialRate的合理值。比如压测得到作业每秒稳定处理1000条，消费的主题有5个分区，那么可以设为1000/5 * 0.8 = 160条/秒/分区。

问题3：开启背压后是否还需要手动配置两个上限参数？ #

需要配置，且属于生产环境必选项。背压的动态调整是基于上一批次的指标计算的，本身存在调整滞后性，如果没有上限约束，遇到突发流量峰值时，首次拉取的消息量可能直接超过作业处理极限，导致作业卡顿甚至崩溃。这两个参数相当于安全闸门，把突发流量限制在作业可承载的范围内，避免意外故障。

问题4：背压生效时spark.streaming.backpressure.initialRate无作用的说法是否正确？ #

该说法错误。initialRate的生效时机仅限于作业首次启动、没有任何历史微批次处理指标的第一个批次，背压启动后的第一个批次会直接用这个参数作为初始拉取速率；等第一个批次处理完成，拿到了处理时间、调度延迟等指标后，背压才会开始动态调整速率，此时initialRate才会失效。它只是生效时间短，并非完全没有作用。

Structured Streaming对应配置补充 #

你当前使用的Structured Streaming没有上述spark.streaming.*的参数，对应逻辑的配置如下：

• 单批次拉取上限：你代码中已经配置的maxOffsetsPerTrigger就是全局的单批次拉取消息总数上限，作用和上述两个上限参数类似
• 分区级拉取上限：通过Kafka消费者参数kafka.max.poll.records配置，限制单次poll每个分区最多拉取的消息数
• 初始速率控制：Structured Streaming无需单独配置初始速率，maxOffsetsPerTrigger本身就会限制首次拉取的消息量，不会出现首次拉取过载的问题

内容的提问来源于stack exchange，提问作者Metadata