反压机制：
spark1.5以后，通过动态收集系统的一些数据来自动的适配集群数据处理能力
在默认情况下，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.kafka.maxRatePerPartition：设定对目标topic每个partition每秒钟拉取的数据条数。 |
|---|
| 假设bacth interval =10s ，maxRatePerPartition设置为2，该topic设置的分区为3 |
| 则一个批次里拉取的总条数=2*3*10=60条数据，也就是当前的吞吐量 |
这种方法虽然可以通过限制接收速率，来适配当前的处理能力，但这种方式存在以下几个问题：
● 我们需要事先估计好集群的处理速度以及消息数据的产生速度；
● 这两种方式需要人工参与，修改完相关参数之后，我们需要手动重启 Spark Streaming 应用程序；
● 如果当前集群的处理能力高于我们配置的 maxRate，而且 producer 产生的数据高于 maxRate，这会导致集群资源利用率低下，而且也会导致数据不能够及时处理。
● 那么有没有可能不需要人工干预，Spark Streaming 系统自动处理这些问题呢？当然有了！Spark 1.5 引入了反压（Back Pressure）机制，其通过动态收集系统的一些数据来自动地适配集群数据处理能力。详细的记录请参见 SPARK-7398 里面的说明。
Spark Streaming 1.5 以前的体系结构
在 Spark 1.5 版本之前，Spark Streaming 的体系结构如下所示：
● 数据是源源不断的通过 receiver 接收，当数据被接收后，其将这些数据存储在 Block Manager 中；为了不丢失数据，其还将数据备份到其他的 Block Manager 中；
● Receiver Tracker 收到被存储的 Block IDs，然后其内部会维护一个时间到这些 block IDs 的关系；
● Job Generator 会每隔 batchInterval 的时间收到一个事件，其会生成一个 JobSet；
● Job Scheduler 运行上面生成的 JobSet。
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 的值。
详细的过程如下图所示：
如何启用？
启用反压：sparkConf.set(“spark.streaming.backpressure.enabled”, “true”)。spark会在作业执行结束后，调用RateController.onBatchCompleted更新batch的元数据信息：batch处理结束时间、batch处理时间、调度延迟时间、batch接收到的消息量等.
在 Spark 启用反压机制很简单，只需要将 spark.streaming.backpressure.enabled 设置为 true 即可，这个参数的默认值为 false。反压机制还涉及以下几个参数，包括文档中没有列出来的：
● spark.streaming.backpressure.initialRate： 启用反压机制时每个接收器接收第一批数据的初始最大速率。默认值没有设置。
● spark.streaming.backpressure.rateEstimator：速率估算器类，默认值为 pid ，目前 Spark 只支持这个，大家可以根据自己的需要实现。
● spark.streaming.backpressure.pid.proportional：用于响应错误的权重（最后批次和当前批次之间的更改）。默认值为1，只能设置成非负值。weight for response to “error” (change between last batch and this batch)
● spark.streaming.backpressure.pid.integral：错误积累的响应权重，具有抑制作用（有效阻尼）。默认值为 0.2 ，只能设置成非负值。weight for the response to the accumulation of error. This has a dampening effect.
● spark.streaming.backpressure.pid.derived：对错误趋势的响应权重。 这可能会引起 batch size 的波动，可以帮助快速增加/减少容量。默认值为0，只能设置成非负值。weight for the response to the trend in error. This can cause arbitrary/noise-induced fluctuations in batch size, but can also help react quickly to increased/reduced capacity.
● spark.streaming.backpressure.pid.minRate：可以估算的最低费率是多少。默认值为 100，只能设置成非负值。
原文链接：https://blog.csdn.net/java_soldier/article/details/83384087
设置spark.streaming.kafka.maxRatePerPartition的原则
spark.streaming.kafka.maxRatePerPartition这个参数是控制吞吐量的，一般和spark.streaming.backpressure.enabled=true一起使用。那么应该怎么算这个值呢？
如例要10分钟的吞吐量控制在5000,0000，kafka分区是10个。
| spark.streaming.kafka.maxRatePerPartition的值 * kafka分区数 * (10 *60)(每秒时间) |
|---|
| 50000000/10/600s =8400 |
也就是我们该设置maxrRatePerPartition这个参数为8400，每秒拉取8400条数据。
spark.streaming.kafka.maxRatePerPartition控制spark读取的每个分区最大消息数。从上面的分析过程可以预见到，每个分区接收到的消息量<=batchDuration * spark.streaming.kafka.maxRatePerPartition.
以下两种场景需要启用反压，可以有效防止应用程序过载：
1、首次启动Streaming应用，kafka保留了大量未消费历史消息，并且auto.offset.reset=latest，可以防止第一个batch接收大量消息、处理时间过长和内存溢出
2、防止kafka producer突然生产大量消息，一个batch接收到大量数据，导致batch之间接收到的数据倾斜
Spark Streaming是先从broker里查询到每个分区的latestOffset，这样就可以得到每个分区的offset range，再用range和上一步预估的速率做对比计算就可以确定每个分区的处理的消息量。整个计算步骤：
1、offset range的消息量 totalLag
2、有效速率=取设置的maxRatePerPartition和预估的速率最小值
3、一个batch的每个分区每秒接收到的消息量=batchDuration*有效速率
offset管理
enable.auto.commit参数必须设置false，因为在自动commit的情况下，可能在一个batch内的数据还没有处理完、或者处理失败，但offset就自动提交了，就会导致数据丢失。下面是在zk中管理offset的思路，zk简单方便而且保证了可用性。
在spark Streaming作业开始时，readOffsets函数用于从zk读取上次应用保存的最后处理的消息偏移量，有以下两种不同处理场景：
1、Spark Streaming应用程序首次运行时，从zk read不到数据，那么就创建一个KafkaConsumer对象，用consumer.position的方式获取offset，这时获取到的offset取决于auto.offset.reset参数的设置
整体：