实时计算引擎Spark笔试题:Spark Streaming 反压机制(Back Pressure)

我就是我 2023-10-06 15:44 20阅读 0赞

Spark Streaming 反压机制是1.5版本推出的特性，用来解决处理速度比摄入速度慢的情况，简单来讲就是做流量控制。当批处理时间(Batch Processing Time)大于批次间隔(Batch Interval，即BatchDuration)时,说明处理数据的速度小于数据摄入的速度，持续时间过⻓或源头数据暴增，容易造成 数据在内存中堆积，最终导致Executor OOM。反压就是来解决这个问题的。  
spark streaming的消费数据源方式有两种：  
1. 若是基于Receiver的数据源，可以通过设置**spark.streaming.receiver.maxRate**来控制最大输入 速率；  
2. 若是基于Direct的数据源(如Kafka Direct Stream)，则可以通过设置spark.streaming.kafka.maxRatePerPartition来控制最大输入速率。

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

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

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

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

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

第二:RateEstimator  
Spark 2.x 只支持基于 PID 的速率估算器，这里只讨论这种实现。基于 PID 的速率估算器简单地说就是它把收集到的数据（当前批次速率）和一个设定值（上一批次速率）进行比较，然后用它们 之间的差计算新的输入值，估算出一个合适的用于下一批次的流量阈值。这里估算出来的值就是流 量的阈值，用于更新每秒能够处理的最大记录数

第三:RateLimiter

RateController和RateEstimator组件都是在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)。同时，需要
    注意，该参数也代表了整个应用生命周期中的最大速率，即使是背压调整的最大值也不会超过该参数。

反压相关的参数

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