火花流Kafka直接消费者消费速度下降

Question

我使用的是运行在AWS中的spark独立集群（spark and spark-streaming-kafka version 1.6.1)，并对检查点目录StreamingContext.getorCreate(config.sparkConfig.CheckpointDir,createStreamingContext)使用S3桶，每个工作节点上没有调度延迟和足够的磁盘空间。

没有更改任何Kafka客户端初始化参数，非常肯定Kafka的结构没有更改：

val kafkaParams = Map("metadata.broker.list" -> kafkaConfig.broker)
val topics = Set(kafkaConfig.topic)
val stream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
  ssc, kafkaParams, topics)

也不明白为什么当直接使用者描述说消耗的偏移量是由流本身消耗的时，我仍然需要在创建流上下文时使用检查点目录？

Answer 1

这通常是通过将spark.streaming.backpressure.enabled设置为true来启用背压的结果。通常情况下，当背压算法看到有更多的数据进入比它所习惯的，它开始将每一批限制到一个相当小的大小，直到它能够重新“稳定”自己。这有时会出现误报，并导致流的处理速度减慢。

如果您想稍微调整一下启发式，它使用了一些未记录的标志（只要确保您知道自己在做什么）：

val proportional = conf.getDouble("spark.streaming.backpressure.pid.proportional", 1.0)
val integral = conf.getDouble("spark.streaming.backpressure.pid.integral", 0.2)
val derived = conf.getDouble("spark.streaming.backpressure.pid.derived", 0.0)
val minRate = conf.getDouble("spark.streaming.backpressure.pid.minRate", 100)

如果您想要血淋淋的细节，那么pidrateestimator就是您需要的。