Spark Streaming 数据限流简述

速率限制的计算与更新

Spark Streaming的流处理其实是基于微批处理(MicroBatch)的，也就是说将数据流按某比较小的时间间隔将数据切割成为一段段微批数据进行处理；

添加监听器

StreamingContext调用Start()启动的时候会将速率控制器（rateController）添加到StreamingListener监听器中；

当每批次处理完成时将触发监听器（RateController），使用该批处理的处理结束时间、处理延迟时间、调度延迟时间、记录行数调用PIDRateEstimator传入PID算法中（PID Controller）计算出该批次的速率（rate）并更新速率限制（rateLimit）与发布该限制速率；

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

for {
  processingEnd <- batchCompleted.batchInfo.processingEndTime
  workDelay <- batchCompleted.batchInfo.processingDelay
  waitDelay <- batchCompleted.batchInfo.schedulingDelay
  elems <- elements.get(streamUID).map(_.numRecords)
 } computeAndPublish(processingEnd, elems, workDelay, waitDelay)
}

private def computeAndPublish(time: Long, elems: Long, workDelay: Long, waitDelay: Long): Unit =
Future[Unit] {
  val newRate = rateEstimator.compute(time, elems, workDelay, waitDelay)
  newRate.foreach { s =>
    rateLimit.set(s.toLong)
    publish(getLatestRate())
  }
}

Socket数据源限流

批次的限制速率上面已经算出，这里说的是接收Socket过来的数据时的数据限流；

SocketInputStream类receive方法接收到数据后将数据存入 BlockGenerator的Buffer中，在写入Buffer前调用限流器 （RateLimiter）对写入数据进行限流；

RateLimiter限流器使用了Google开源的 Guava中内置的RateLimiter限流器，该类只是对Guava限流器的简单封装；

在Spark Streaming中可通过使用两个参数配置初始速率与最大速率spark.streaming.receiver.maxRate、spark.streaming.backpressure.initialRate；亦可配置PIDController算法相关的四个参数值；

RateLimiter限流器是基于令牌桶的算法基本原理比较简单，以一个恒定的速率生成令牌放入令牌桶中，桶满则停止，处理请求时需要从令牌桶中取出令牌，当桶中无令牌可取时阻塞等待，此算法用于确保系统不被洪峰击垮。

private lazy val rateLimiter = GuavaRateLimiter.create(getInitialRateLimit().toDouble)
/**
  * Push a single data item into the buffer.
 */
 def addData(data: Any): Unit = {
  if (state == Active) {
//调用限流器等待
  waitToPush()
  synchronized {
    if (state == Active) {
      currentBuffer += data
    } else {
      throw new SparkException(
        "Cannot add data as BlockGenerator has not been started or has been stopped")
    }
  }
} else {
  throw new SparkException(
    "Cannot add data as BlockGenerator has not been started or has been stopped")
 }
}

def waitToPush() {
   //限流器申请令牌
   rateLimiter.acquire()
}

Guava库中RateLimiter限流器基本使用：

//创建限流器，每秒产生令牌数1
    RateLimiter rateLimiter=RateLimiter.create(1);
    for (int i = 0; i < 10; i++) {
        //获得一个令牌，未申请到令牌则阻塞等待
        double waitTime = rateLimiter.acquire();
        System.out.println(String.format("id:%d time:%d waitTime:%f",i,System.currentTimeMillis(),waitTime));
    }

Kafka数据源限流的实现

在Spark Streaming Kafka包拉取Kafka数据会进行如下动作：

1、取Kafka中最新偏移量、分区

2、通过rateController限制每个分区可拉取的最大消息数

3、在DirectKafkaInputDStream中创建KafkaRDD，在其中调用相关对象拉取数据

通过如上步骤也可用看出，只要限制了Kafka某个分区的偏移量（offset）范围也就可限制从Kafka拉取的消息数量，从而达到限流的目的，Spark streaming kafka也是通过此实现的；

计算每个分区速率限制，有如下步骤：

1、通过seekToEnd获取最新可用偏移量与当前偏移量对比获得当前所有分区延迟偏移量

单个分区偏移量延迟=最新偏移量记录-当前偏移量记录

2、获取配置项中每个分区最大速率 （spark.streaming.kafka.maxRatePerPartition），背压率计算，计算每个分区背压率计算公式为：

单个分区背压率=单个分区偏移量延迟/所有分区总延迟*速率限制

速率限制（rateLimit）：为通过PIDController动态计算得来

如有配置 每个分区最大速率 则取配置项最大速率与 背压率 两者中的 最小值 ， 未配置 则取 背压率 作为每个分区速率限制；

3、将批次间隔（batchDuration）*每个分区速率限制=每个分区最大消息数

4、取当前分区偏移量+分区最大消息数 与 最新偏移量两者当中最小的，由此来控制拉取消息速率；

如当前偏移量+分区最大消息数 大于 最新偏移量则取 最新偏移量否则取 当前偏移量+分区最大消息数作为 拉取Kafka数据的Offset范围 ；

// 限制每个分区最大消息数
protected def clamp(
offsets: Map[TopicPartition, Long]): Map[TopicPartition, Long] = {

maxMessagesPerPartition(offsets).map { mmp =>
  mmp.map { case (tp, messages) =>
      val uo = offsets(tp)
      tp -> Math.min(currentOffsets(tp) + messages, uo)
  }
}.getOrElse(offsets)
}

参考资料：

http://kafka.apache.org

http://spark.apache.org