apache-kafka - Spark Streaming spark.streaming.backpressure.pid.minRate 是每个分区还是每个批处理间隔的消息总数？

Question

我有一个 Spark 流应用程序，它使用 Spark Direct Streaming(不是接收器)方法从 Kafka 读取消息并按分区处理消息。

在我的 Kafka 分区中，有时我们得到的消息需要 20 秒来处理 2000 条消息，而一些消息需要 7-9 秒才能处理相同的编号。的消息。

鉴于波动，我们开启背压设置如下。

spark.batch.duration=10 seconds
spark.streaming.kafka.maxRatePerPartition=200

spark.streaming.backpressure.enabled=true
spark.streaming.backpressure.initialRate=60
spark.streaming.kafka.maxRatePerPartition=200

并且还使用以下参数指定了 RateEstimator。我不了解 PID 的数学原理，但尝试了不同的组合，其中之一如下。

spark.streaming.backpressure.rateEstimator=pid
spark.streaming.backpressure.pid.minRate=1600
spark.streaming.backpressure.pid.integral=1
spark.streaming.backpressure.pid.proportional=25
spark.streaming.backpressure.pid.derived=1

最初，spark 读取 RDD 中 1 个分区的 2000 条消息，但一段时间后它开始读取 800 条记录。我认为是 minRate/2。然后它保持静止..在日志中，它始终打印 1600 作为新汇率。

2017-01-20 14:55:14 TRACE PIDRateEstimator:67 - New rate = 1600.0

鉴于我的情况，我有几个问题:

1. 是 spark.streaming.backpressure.pid.minRate 每个分区还是总数。要批量阅读的邮件数量？
2. 为什么要阅读 800 条消息而不是 1600 条？
3. 是否有任何建议的参数可以在处理时间较长时降低输入速率并在处理速度非常快时增加回接近 maxRatePerPartition 的值？在我的示例中，输入速率从 2000 开始，但当平均花费 20 秒左右时，它会减少到 800，但当 800 条消息在 3-4 秒内处理时，它不会增加到 1600 或更多。这会导致时间浪费和低吞吐量。

Answer 1

第 1 个问题的答案 -spark.streaming.backpressure.pid.minRate 是每个分区每秒的消息数。

关于您的消息消费率的差异，可能是由于pid配置错误。

在大多数情况下，pid.proportional、pid.integral 和 pid.derived 的默认值就足够了。因此，如果任何人对 pid 速率估计器背后的数学没有正确的理解，建议继续使用默认值。我要调整的唯一参数是 spark.streaming.backpressure.pid.minRate

参见:https://richardstartin.github.io/posts/tuning-spark-back-pressure-by-simulation https://www.linkedin.com/pulse/enable-back-pressure-make-your-spark-streaming-production-lan-jiang/