spark的反压与推测机制

反压背景

在默认情况下，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.consumer.poll.ms：设置spark拉取kafka数据的等待时间，默认 512毫秒 太短需要设置，这种方法虽然可以通过限制接收速率，来适配当前的处理能力，但这种方式存在以下几个问题：

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

Spark Streaming反压机制的核心原理

简单介绍：反压机制核心功能就是控制spark程序最多能消费的数据量，保证数据不积压

注意事项：一旦启动反压机制，设置spark.streaming.kafka.maxRatePerPartition kafka最大分区拉取数就失效了

Spark Streaming 反压机制的使用

        在 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，只能设置成非负值。

推测机制

spark推测式执行

            推测任务是指对于一个Stage里面拖后腿的Task，会在其他节点的Executor上再次启动这个task，如果其中一个Task实例运行成功则将这个最先完成的Task的计算结果作为最终结果，同时会干掉其他Executor上运行的实例。spark推测式执行默认是关闭的，可通过spark.speculation属性来开启。

推测机制原理

           当成功的Task数超过总Task数的75%(可通过参数spark.speculation.quantile设置进行更改)时，再统计所有成功的Tasks的运行时间，得到一个中位数，用这个中位数乘以1.5(可通过参数spark.speculation.multiplier控制进行更改)得到运行时间门限，如果在运行的Tasks的运行时间超过这个门限，则对它启用推测。简单来说就是对那些拖慢整体进度的Tasks启用推测，以加速整个Stage的运行

// 数据出现某个task执行不完的情况，开启推测机制，有可能是网络问题等等
		/*conf.set("spark.speculation.interval","100") // 检测周期，单位为毫秒
		conf.set("spark.speculation.quantile","0.75") // 完成task的百分比时启动推测
		conf.set("spark.speculation.multiplier","1.5") // 比其它的慢多少倍时启动推测
		conf.set("spark.streaming.concurrentJobs","3") // 控制job并发度，启动线程执行*/

扫一扫加入大数据技术交流群，了解更多大数据技术，还有免费资料等你哦

扫一扫加入大数据技术交流群，了解更多大数据技术，还有免费资料等你哦

扫一扫加入大数据技术交流群，了解更多大数据技术，还有免费资料等你哦