Spark Streaming Backpressure 分析

一般默认情况下，Spark Streaming 通过 receivers (或 Direct 方式) 以生产者生产数据的速率接收数据。 当 batch processing time > batch interval 的时候，即每个批次数据处理的时间要比 Spark Streaming 批处理间隔时间长；

一、介绍

1、生产中可能遇到的情况是什么？

一般默认情况下，Spark Streaming 通过 receivers (或 Direct 方式) 以生产者生产数据的速率接收数据。
当 batch processing time > batch interval 的时候，即每个批次数据处理的时间要比 Spark Streaming 批处理间隔时间长；

越来越多的数据被接收，但是数据的处理速度没有跟上，则导致系统开始出现数据堆积，而可能导致 Executor 端出现 OOM 问题而出现失败的情况。

在 Spark 1.5 版本之前，为了解决这个问题，对于 Receiver-based 数据接收器，我们可以通过配置 spark.streaming.receiver.maxRate 参数来限制每个 receiver 每秒最大可以接收的记录的数据；
对于 Direct Approach 的数据接收，我们可以通过配置 spark.streaming.kafka.maxRatePerPartition 参数来限制每次作业中每个 Kafka 分区最多读取的记录条数。但是这种方法虽可以通过限制接收速率，来适配当前的处理能力，但这种方式存在以下几个问题：

• 需要人为事先估计好集群的处理速度以及消息数据的产生速度
  
• 修改完相关参数之后，需手动重启 Spark Streaming 应用程序
  
• 如果当前集群的处理能力高于我们配置的 maxRate，且 producer产生的数据的速度高于maxRate，则导致集群资源利用率低下、数据不能够及时处理
  

  2、什么是反压机制？

  在Spark 1.5 中引入了反压（Back Pressure）机制，通过动态收集系统的一些数据来自动地适配集群数据处理能力。

(1)Spark Streaming 1.5 以前的体系结构

数据是源源不断的通过 receiver 接收，当数据被接收后，其将这些数据存储在 Block Manager 中；
为了不丢失数据，其还将数据备份到其他的 Block Manager 中；
Receiver Tracker 收到被存储的 Block IDs，然后其内部会维护一个时间到这些 block IDs 的关系；
Job Generator 会每隔 batchInterval 的时间收到一个事件，其会生成一个 JobSet；
Job Scheduler 运行上面生成的 JobSet。

(2)Spark Streaming 1.5 之后的体系结构

为了实现自动调节数据的传输速率，在原有的架构上新增了一个名为 RateController 的组件，这个组件继承自 StreamingListener，其监听所有作业的 onBatchCompleted 事件，并基于 processingDelay 、schedulingDelay 、当前 Batch 处理的记录条数以及处理完成事件来估算出一个速率；这个速率主要用于更新流每秒能够处理的最大记录的条数。

速率估算器（RateEstimator）可以又多种实现，不过目前的 Spark 2.2 只实现了基于 PID 的速率估算器。

InputDStreams 内部的 RateController 里面会存下计算好的最大速率，这个速率会在处理完 onBatchCompleted 事件之后将计算好的速率推送到 ReceiverSupervisorImpl，这样接收器就知道下一步应该接收多少数据了。

如果用户配置了 spark.streaming.receiver.maxRate 或 spark.streaming.kafka.maxRatePerPartition，那么最后到底接收多少数据取决于三者的最小值。也就是说每个接收器或者每个 Kafka 分区每秒处理的数据不会超过 spark.streaming.receiver.maxRate 或 spark.streaming.kafka.maxRatePerPartition 的值。

详细的过程如下图：

3、怎样使用Spark Streaming 反压机制？

Spark 启用反压机制： spark.streaming.backpressure.enabled=true (默认值为 false)
反压机制还涉及以下几个参数，但是有部分参数在官方文档中没有列出来，因为在不同版本的spark中不建议设置，涉及的参数：

(1)spark.streaming.backpressure.initialRate
启用反压机制时每个接收器接收第一批数据的初始最大速率
(2)spark.streaming.backpressure.rateEstimator
速率估算器类，默认值为 pid
(3)spark.streaming.backpressure.pid.integral
错误积累的响应权重，具有抑制作用（有效阻尼）。默认值为 0.2 ，只能设置为0或者大于0的数
(4)spark.streaming.backpressure.pid.derived
对错误趋势的响应权重。 这可能会引起 batch size 的波动，可以帮助快速增加/减少容量。默认值为0.0，只能设置为0或者大于0的数
(5)spark.streaming.backpressure.pid.minRate
可以估算的最低费率是多少。默认值为 100，只能设置为0或者大于0的数
(6)spark.streaming.backpressure.pid.proportional
用于响应错误的权重（最后批次和当前批次之间的更改）。默认值为1.0，只能设置为0或者大于0的数

注意，以上参数可以在文档中寻找到

• 文档 1
• 文档 2

二、参考说明

1、该功能自spark-1.5.0版本后有，发行说明：文档 3

2、对应的Spark提交文档：文档 4 文档 5

3、文档参考：

• 文档 6
• 文档 7
• 文档 8
• 文档 9