Spark Streaming反压机制

反压(Back Pressure)机制主要用来解决流处理系统中,处理速度比摄入速度慢的情况。是控制流处理中批次流量过载的有效手段。

1 反压机制原理

Spark Streaming中的反压机制是Spark 1.5.0推出的新特性,可以根据处理效率动态调整摄入速率。

1.1 反压定义

当批处理时间(Batch Processing Time)大于批次间隔(Batch Interval,即 BatchDuration)时,说明处理数据的速度小于数据摄入的速度,持续时间过长或源头数据暴增,容易造成数据在内存中堆积,最终导致Executor OOM或任务奔溃。

1.2 反压的数据源方式及限流处理

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会自动根据处理能力来调整输入速率,从而在流量高峰时仍能保证最大的吞吐和性能。

1.3 反压的实现原理

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

  • RateController
  • RateEstimator
  • RateLimiter

主要是通过RateController组件来实现。RateController继承自接口StreamingListener,并实现了onBatchCompleted方法。每一个Batch处理完成后都会调用此方法,具体如下:

override def onBatchCompleted(batchCompleted: StreamingListenerBatchCompleted) {
    val elements = batchCompleted.batchInfo.streamIdToInputInfo

    for {
      // 处理结束时间
      processingEnd  batchCompleted.batchInfo.processingEndTime
      // 处理时间,即processingEndTime - processingStartTime
      workDelay  batchCompleted.batchInfo.processingDelay
      // 在调度队列中的等待时间,即processingStartTime - submissionTime
      waitDelay  batchCompleted.batchInfo.schedulingDelay
      // 当前批次处理的记录数
      elems  elements.get(streamUID).map(_.numRecords)
    } computeAndPublish(processingEnd, elems, workDelay, waitDelay)
  }

可以看到,接着又调用的是 computeAndPublish方法,如下:

private def computeAndPublish(time: Long, elems: Long, workDelay: Long, waitDelay: Long): Unit =
    Future[Unit] {
      // 根据处理时间、调度时间、当前Batch记录数,预估新速率
      val newRate = rateEstimator.compute(time, elems, workDelay, waitDelay)
      newRate.foreach { s =>
      // 设置新速率
        rateLimit.set(s.toLong)
      // 发布新速率
        publish(getLatestRate())
      }
    }

更深一层,具体调用的是 rateEstimator.compute方法来预估新速率,

def compute(
      time: Long,
      elements: Long,
      processingDelay: Long,
      schedulingDelay: Long): Option[Double]

RateEstimator是速率估算器,主要用来估算最大处理速率,默认的在2.2之前版本中只支持PIDRateEstimator,在以后的版本可能会支持使用其他插件,其源码如下:

def create(conf: SparkConf, batchInterval: Duration): RateEstimator =
    conf.get("spark.streaming.backpressure.rateEstimator", "pid") match {
      case "pid" =>
        val proportional = conf.getDouble("spark.streaming.backpressure.pid.proportional", 1.0)
        val integral = conf.getDouble("spark.streaming.backpressure.pid.integral", 0.2)
        val derived = conf.getDouble("spark.streaming.backpressure.pid.derived", 0.0)
        val minRate = conf.getDouble("spark.streaming.backpressure.pid.minRate", 100)
        new PIDRateEstimator(batchInterval.milliseconds, proportional, integral, derived, minRate)
    //默认的只支持pid,其他的配置抛出异常
      case estimator =>
        throw new IllegalArgumentException(s"Unknown rate estimator: $estimator")
    }

以上这两个组件都是在Driver端用于更新最大速度的,而RateLimiter是用于接收到Driver的更新通知之后更新Executor的最大处理速率的组件。RateLimiter是一个抽象类,它并不是Spark本身实现的,而是借助了第三方Google的GuavaRateLimiter来产生的。

它实质上是一个限流器,也可以叫做令牌,如果Executor中task每秒计算的速度大于该值则阻塞,如果小于该值则通过,将流数据加入缓存中进行计算。这种机制也可以叫做令牌桶机制,图示如下:

Spark Streaming反压机制

接收到的newRate进行比较,取两者中的最小值来作为最大处理速率,如果没有设置,直接设置为newRate。源码如下:

private[receiver] def updateRate(newRate: Long): Unit =
    if (newRate > 0) {
      if (maxRateLimit > 0) {
        //如果设置了maxRateLimit则取两者中的最小值
        rateLimiter.setRate(newRate.min(maxRateLimit))
      } else {
        rateLimiter.setRate(newRate)
      }
    }

spark 1.5引入的反压机制架构图如下:

Spark Streaming反压机制

2. 反压机制相关参数

参数名称 默认值 说明 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 只能为正数,最小速率

3. 反压机制的使用

//启用反压
conf.set("spark.streaming.backpressure.enabled","true")
//最小摄入条数控制
conf.set("spark.streaming.backpressure.pid.minRate","1")
//最大摄入条数控制
conf.set("spark.streaming.kafka.maxRatePerPartition","12")

注意:

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

4. 查看日志

创建速率控制器
INFO PIDRateEstimator: Created PIDRateEstimator with proportional = 1.0, integral = 0.2, derivative = 0.0, min rate = 1.0
计算当前批次速率
// records 记录数(对应WebUI: Input Size)
// processing time 处理时间,毫秒(对应WebUI: Processing Time)
// scheduling delay 调度时间,毫秒(对应WebUI: Scheduling Delay)
TRACE PIDRateEstimator:
time = 1558888897224, # records = 33, processing time = 24548, scheduling delay = 8
预估新批次速率
TRACE PIDRateEstimator:
 latestRate = -1.0, error = -2.344305035033404
 latestError = -1.0, historicalError = 0.0010754440280267231
 delaySinceUpdate = 1.558888897225E9, dError = -8.623482003280801E-10
第一次计算跳过速率估计
TRACE PIDRateEstimator: First run, rate estimation skipped
当前批次没有记录或没有延迟则跳过速率估计
TRACE PIDRateEstimator: Rate estimation skipped
以新的预估速率运行
TRACE PIDRateEstimator: New rate = 1.0

WebUI

Spark Streaming反压机制

可以看到,开启反压后,摄入速率 Input Rate可以根据处理时间 Processing Time来调整,从而保证应用的稳定性。

Original: https://www.cnblogs.com/lenmom/p/12022277.html
Author: 老董
Title: Spark Streaming反压机制

原创文章受到原创版权保护。转载请注明出处:https://www.johngo689.com/8840/

转载文章受原作者版权保护。转载请注明原作者出处!

(0)

大家都在看

发表回复

登录后才能评论
免费咨询
免费咨询
扫码关注
扫码关注
联系站长

站长Johngo!

大数据和算法重度研究者!

持续产出大数据、算法、LeetCode干货,以及业界好资源!

2022012703491714

微信来撩,免费咨询:xiaozhu_tec

分享本页
返回顶部