Spark Streaming 反压机制

反压机制的引入

默认情况下，Spark Streaming通过receiver以生产者生产数据的速率接收数据，计算过程中会出现批处理时间（Processing Time）大于批间隔时间（Batch Interval）的情况，其中批处理时间为实际计算一个批次花费的时间，批间隔时间为Streaming应用设置的批处理间隔。这意味着Spark Streaming的数据接收速率高于Spark从队列中移除数据的速率，即处理能力低，在设置间隔内不能完全处理当前接收速率接收的数据。如果这种情况持续过长的时间，会造成数据在内存中堆积，导致receiver所在excutor内存溢出等问题（如果设置StorageLevel包含disk，则内存存放不下的数据会溢写至disk，加大延时）。在Spark1.5以前的版本中，用户如果要限制receiver的数据接收速率，可以通过设置静态配置参数：

spark.streaming.receiver.maxRate

的值来实现。此举虽然可以通过限制接收速率来适配当前的处理能力、防止内存溢出，但也会引入其他问题。比如，producer数据生产高于maxRate，当前集群处理能力也高于maxRate，这就会造成资源利用率下降等问题。为了更好地协调数据接收速率与资源处理能力，Spark Streaming从v1.5开始引入动态反压机制，通过动态控制数据接收速率来适配集群数据处理能力。

简单易懂的话描述反压机制

比如现在有个application，5s接收5000条数据是最优的情况，但是现实情况是5s接收了10000条数据，processing time 假设为30s，反压机制会根据Batch Interval的时间，动态调节接收的数据，它发现5s接收的数据，30s才处理完成，这样下去会有任务堆积，它会把10000条数据逐渐动态调整减少，比如先调整为8000条，发现Processing Time为20s，然后再继续调整为6000条，发现Processing Time为8s以此类推，直到Active Batches里面没有任务堆积，也就是Processing Time为5s，即接收数据的时间和Processing Time相同为止。

为什么官网说最好和spark.streaming.receiver.maxRate还有spark.streaming.kafka.maxRatePerPartition这两个参数一起设置最好呢？

如果5s接收的不是10000条，而是接收了100000条数据，还是5000条是个正常水平的话，那么等调节到5000条的时候，可能已经堆积了太多的任务了。这样如果内存不足或磁盘不足直接会导致集群崩掉了。因此可以通过上面的参数指定一次最多接收5500条，那么系统调节到5000的水平将会很快，也就不会有大量任务堆积的问题。