——————————————————————————————
Spark Streaming概述
Spark Streaming 初始化过程
Spark Streaming Receiver启动过程分析
Spark Streaming 数据准备阶段分析（Receiver方式）
Spark Streaming 数据计算阶段分析
SparkStreaming Backpressure分析
Spark Streaming Executor DynamicAllocation 机制分析

——————————————————————————————

1、 引入Streaming Executor DynamicAllocation 机制的原因

在Spark Streaming，作业的执行是以批处理的方式进行的，批处理间隔内(batch interval)要完成对批量作业的执行，这就要求作业的执行时间（process time）不大于设定的批处理间隔。在计算资源一定的情况下，执行时间与待处理的数据规模成正比。在大数据流式计算环境中，数据的产生完全由数据源决定,由于不同的数据源在不同时空范围内的状态不统一且发生动态变化，导致数据流的速率呈现出了突发性的特征。前一时刻数据速率和后一时刻数据速率可能会有巨大的差异， 因此不同批次接收的数据总量存在差异，不同批次的执行时间也就会存在差异。这种差异是由于计算资源与数据规模不匹配造成的。在Spark 2.0以前， Streaming的资源分配都采取资源预先分配的策略，资源管理器依据应用的申请量予以提前分配，在应用执行期间不能依据应用实际的计算资源需求量进行调整，执行作业时会存在资源过剩或资源不足的情况。

2、Spark Streaming 的计算特征

Spark Streaming应用的执行过程可以分成数据准备和数据计算两个阶段，两个阶段分别由不同的作业进行处理，分别是数据接收作业（提交Receiver时的Job）和数据处理作业(Batch Job)。在大数据流式计算中，数据是实时产生、动态增加的，只要数据源处于活动状态,数据就会一直产生，因此要求数据接收作业不间断的接收数据，并将接收的数据分割成批，供数据处理作业消费。数据处理作业以批处理方式运行。一般情况下，批处理的执行时间不大于其批处理间隔，一个批次可以分成作业实际处理阶段和等待下一个批次阶段。综上所述，Spark Streaming的计算特征如下图所示（实际处理时间<批处理间隔）:

SparkStreaming 计算特征

另外，特殊情况下，会存在实际处理阶段所用的时间 > 批处理时间。这种情况下，一般是由于作业处理能力弱引起的。

3、Streaming Executor DynamicAllocation机制的评价指标

经过上述分析，作业的实际处理时间与设定的批处理间隔之间存在的关系如下：

• 作业的实际处理时间 远小于 批处理间隔
  此时，计算资源大部分时间处于空闲，造成资源浪费
• 作业的实际处理时间 约等于 批处理间隔
  此时，能满足处理，但数据具有波动性，可能下一个时间不能满足，造成延迟。计算资源以初显不足。
• 介于以上两者之间
  系统资源能正常满足计算，又不至于造成过多浪费，此时系统稳定性良好。
• 作业的实际处理时间 大于 批处理间隔
  批处理间隔内不能完成批作业，说明计算资源不足。

因此，采用 有效处理时间占比【有效处理时间占比 = 实际处理时间 / 批处理间隔】来评价当前计算资源的过剩或不足。

4 、Streaming Executor DynamicAllocation机制的工作流程

• 通过监控组件，获取已完成批作业的实际执行时间
• 计算有效处理时间占比，然后与设置的阈值进行比较，如果小于下限down,则结束一个Executor; 如果大于上限up， 则增加Executor(个数由比率决定).

该功能启用与否由参数spark.streaming.dynamicAllocation.enabled参数决定，默认为false不开启。

5、源码分析

与Spark Streaming Backpressure机制相同，Spark Streaming Executor动态伸缩机制也是以事件驱动的形式工作的。其负责动态伸缩的类为ExecutorAllocationManager，其继承自StreamingListener。 类定义结构如下：

private[streaming] class ExecutorAllocationManager(
    client: ExecutorAllocationClient,
    receiverTracker: ReceiverTracker,
    conf: SparkConf,
    batchDurationMs: Long,
    clock: Clock) extends StreamingListener with Logging {
    ......
    ......
}

ExecutorAllocationManager 负责管理分配给StreamingContext（Streaming应用）的Executor.其通过分析Streaming作业的监控信息，动态的伸请或释放executor.

5.1 ExecutorAllocationManager 注册与启动

JobScheduler启动时会创建ExecutorAllocationManager 并向ListenerBus注册并开启监听。

  def start(): Unit = synchronized {
    ......

    val executorAllocClient: ExecutorAllocationClient = ssc.sparkContext.schedulerBackend match {
      case b: ExecutorAllocationClient => b.asInstanceOf[ExecutorAllocationClient]
      case _ => null
    }
    ......
    executorAllocationManager = ExecutorAllocationManager.createIfEnabled(
      executorAllocClient,
    executorAllocationManager.foreach(ssc.addStreamingListener)
   ......
    executorAllocationManager.foreach(_.start())
    logInfo("Started JobScheduler")
  }

其中executorAllocationManager的start方法定义如下：

  def start(): Unit = {
    timer.start()
    logInfo(s"ExecutorAllocationManager started with " +
      s"ratios = [$scalingUpRatio, $scalingDownRatio] and interval = $scalingIntervalSecs sec")
  }

  private val timer = new RecurringTimer(clock, scalingIntervalSecs * 1000,
    _ => manageAllocation(), "streaming-executor-allocation-manager")

其将开启定时器，周期性的执行manageAllocation方法，其时间周期由参数 “spark.streaming.dynamicAllocation.scalingInterval”决定，默认为60s.
即：默认情况下每隔60s 执行一次manageAllocation. managerAllocation会使用历史作业执行信息计算出有效处理时间占比ratio, 并依据占比与预设阈值的关系决定增减资源。预设的阈值信息及增减策略为：

• scalingUpRatio
  阈值上限，由参数“park.streaming.dynamicAllocation.scalingUpRatio”控制，默认值为0.9。当计算出的ratio大于scalingUpRatio 时，将按如下算式计算出的值，增加若干Executor。

 math.max(math.round(ratio).toInt, 1)

• scalingDownRatio
  阈值下限， 由参数“spark.streaming.dynamicAllocation.scalingDownRatio”，默认值为0.3。 当计算出的ratio小于scalingDownRatio时，将减少一个Executor.

manageAllocation的实现如下：

/**
   * Manage executor allocation by requesting or killing executors based on the collected
   * batch statistics.
   */
  private def manageAllocation(): Unit = synchronized {
    logInfo(s"Managing executor allocation with ratios = [$scalingUpRatio, $scalingDownRatio]")
    if (batchProcTimeCount > 0) {
      val averageBatchProcTime = batchProcTimeSum / batchProcTimeCount
      val ratio = averageBatchProcTime.toDouble / batchDurationMs
      logInfo(s"Average: $averageBatchProcTime, ratio = $ratio" )
      if (ratio >= scalingUpRatio) {
        logDebug("Requesting executors")
        val numNewExecutors = math.max(math.round(ratio).toInt, 1)
        requestExecutors(numNewExecutors)
      } else if (ratio <= scalingDownRatio) {
        logDebug("Killing executors")
        killExecutor()
      }
    }
    batchProcTimeSum = 0
    batchProcTimeCount = 0
  }

经过分析代码可知，初始情况下，因if条件batchProcTimeCount > 0不满足，上述机制并不会执行。只有onBatchCompleted事件触发之后，即有作业执行完成，可以做为判断依据之后，manageAllocation（）才会正式生效。

5.2 事件触发

Executor dynamicAllocation机制是依据完成的批Job的执行信息进行决策，其在批Job执行完成时会收集作业。
其事件触发过程，同“SparkStreaming Backpressure分析”中"3.3.1 BatchCompleted触发过程"一节过程相同，不再赘述。

5.3 事件处理

onBatchCompleted的事件处理定义如下：

override def onBatchCompleted(batchCompleted: StreamingListenerBatchCompleted): Unit = {
    logDebug("onBatchCompleted called: " + batchCompleted)
    if (!batchCompleted.batchInfo.outputOperationInfos.values.exists(_.failureReason.nonEmpty)) {
      batchCompleted.batchInfo.processingDelay.foreach(addBatchProcTime)
    }
  }

当onBatchCompleted事件处发时，将执行addBatchProcTime, 为batchProcTimeSum及batchProcTimeCount 增值。

 private def addBatchProcTime(timeMs: Long): Unit = synchronized {
    batchProcTimeSum += timeMs
    batchProcTimeCount += 1
    logDebug(
      s"Added batch processing time $timeMs, sum = $batchProcTimeSum, count = $batchProcTimeCount")
  }

5.4 生效时机

当Timer 再次触发时，manageAllocation 的if条件将满足，其会依据决策信息对Executor进行增减操作。 同时重置batchProcTimeSum及batchProcTimeCount的值，开始下一轮次的准备工作。

6、与Spark Core中Executor DynamicAllocation 机制的区别

Spark Core 中的DynamicAllocation机制同Spark Streaming中一样，也是周期性调度评估机制确定是否要进行Executor增减。不同的是，Spark Core中应用是批处理应用，执行完成之后，即可以结束，因此其使用idle策略来评估Executor的增减，具体策略为：

• 减少Executor
  应用拥有的Executor数量在处理当前负载时绰绰有余，通过缩减Executor仍然能够一次性执行所有任务（running + pending task）时，会通过一定的策略结束掉部分Executor，达到节省资源的目的。Spark Core中的策略为如果监测到一个Executor空闲了K 秒，这意味着其在未来不再执行任务，则可以将其移除。其中参数K由“spark.dynamicAllocation.executorIdleTimeout ”配制，默认值为60s.
• 增加Executor
  应用拥有的Executor数量不足以及时处理当前负载，存在任务长时间堆集，则要增加Executor。Spark Core中的策略为如果监测到任务调度队列中的任务N秒内没有被调度，则会增加新的Executor，如果再过M秒还未进行调度，则以指数方式继续增加，直到上限。

由于Spark Streaming应用是长期存在的、微批处理应用。其每隔一个小的时间间隔(batchInterval)就会提交微批处理应用。其会使Spark Core中的Executor DynamicAllocation 机制的idle 策略受阻，因此Spark Streaming 采用有效处理时间占比radio来进行决策。