Spark Streaming 反压机制(Back Pressure)

当前位置：首页 / 技术干货 / 正文

2022-07-29

速率 spark streaming 处理

Spark Streaming 反压机制(Back Pressure)

Spark Streaming 反压机制是1.5版本推出的特性，用来解决处理速度比摄入速度慢的情况，简单来讲就是做流量控制。当批处理时间(Batch Processing Time)大于批次间隔(Batch Interval，即 BatchDuration)时,说明处理数据的速度小于数据摄入的速度，持续时间过长或源头数据暴增，容易造成数据在内存中堆积，最终导致Executor OOM。反压就是来解决这个问题的。

spark streaming的消费数据源方式有两种：

若是基于Receiver的数据源，可以通过设置spark.streaming.receiver.maxRate来控制最大输入速率；若是基于Direct的数据源(如Kafka Direct Stream)，则可以通过设置spark.streaming.kafka.maxRatePerPartition来控制最大输入速率。

当然，在事先经过压测，且流量高峰不会超过预期的情况下，设置这些参数一般没什么问题。但最大值，不代表是最优值，最好还能根据每个批次处理情况来动态预估下个批次最优速率。

在Spark 1.5.0以上，就可通过背压机制来实现。开启反压机制，即设置spark.streaming.backpressure.enabled为true，Spark Streaming会自动根据处理能力来调整输入速率，从而在流量高峰时仍能保证最大的吞吐和性能

Spark Streaming的反压机制中，有以下几个重要的组件：

RateController 组件是 JobScheduler 的监听器，主要监听集群所有作业的提交、运行、完成情况，并从 BatchInfo 实例中获取以下信息，交给速率估算器（RateEstimator）做速率的估算。

当前批次任务处理完成的时间戳（processingEndTime）
该批次从第一个 job 到最后一个 job 的实际处理时长（processingDelay）
该批次的调度时延，即从被提交到 JobScheduler 到第一个 job 开始处理的时长（schedulingDelay）
该批次输入数据的总条数（numRecords）

Spark 2.x 只支持基于 PID 的速率估算器，这里只讨论这种实现。基于 PID 的速率估算器简单地说就是它把收集到的数据（当前批次速率）和一个设定值（上一批次速率）进行比较，然后用它们之间的差计算新的输入值，估算出一个合适的用于下一批次的流量阈值。这里估算出来的值就是流量的阈值，用于更新每秒能够处理的最大记录数以上这两个组件都是在Driver端用于更新最大速度的，而RateLimiter是用于接收到Driver的更新通知之后更新Executor的最大处理速率的组件。RateLimiter是一个抽象类，它并不是Spark本身实现的，而是借助了第三方Google的GuavaRateLimiter来产生的。它实质上是一个限流器，也可以叫做令牌，如果Executor中task每秒计算的速度大于该值则阻塞，如果小于该值则通过，将流数据加入缓存中进行计算。

* 反压机制真正起作用时需要至少处理一个批：由于反压机制需要根据当前批的速率，预估新批的速率，所以反压机制真正起作用前，应至少保证处理一个批。

* 如何保证反压机制真正起作用前应用不会崩溃：要保证反压机制真正起作用前应用不会崩溃,需要控制每个批次最大摄入速率。若为Direct Stream，如Kafka Direct Stream,则可以通过spark.streaming.kafka.maxRatePerPartition参数来控制。此参数代表了每秒每个分区最大摄入的数据条数。假设BatchDuration为10秒,spark.streaming.kafka.maxRatePerPartition为12条,kafka topic 分区数为3个，则一个批(Batch)最大读取的数据条数为360条(3*12*10=360)。同时，需要注意，该参数也代表了整个应用生命周期中的最大速率，即使是背压调整的最大值也不会超过该参数。

反压相关的参数

参数名称	默认值	说明
spark.streaming.backpressure.enabled	false	是否启用反压机制
spark.streaming.backpressure.initialRate	无	初始最大接收速率。只适用于Receiver Stream，不适用于Direct Stream。
spark.streaming.backpressure.rateEstimator	pid	速率控制器,Spark 默认只支持此控制器，可自定义。
spark.streaming.backpressure.pid.proportional	1.0	只能为非负值。当前速率与最后一批速率之间的差值对总控制信号贡献的权重。用默认值即可。
spark.streaming.backpressure.pid.integral	0.2	只能为非负值。比例误差累积对总控制信号贡献的权重。用默认值即可
spark.streaming.backpressure.pid.derived	0	只能为非负值。比例误差变化对总控制信号贡献的权重。用默认值即可
spark.streaming.backpressure.pid.minRate	100	只能为正数，最小速率