第17课:Spark Streaming资源动态申请和动态控制消费速率原理剖析

Posted

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了第17课:Spark Streaming资源动态申请和动态控制消费速率原理剖析相关的知识,希望对你有一定的参考价值。

本期内容:

  • Spark Streaming资源动态分配

  • Spark Streaming动态控制消费速率

为什么需要动态?

  • Spark默认情况下粗粒度的,先分配好资源再计算。而Spark Streaming有高峰值和低峰值,但是他们需要的资源是不一样的,如果按照高峰值的角度的话,就会有大量的资源浪费。

  • Spark Streaming不断的运行,对资源消耗和管理也是我们要考虑的因素。

  • Spark Streaming资源动态调整的时候会面临挑战:

  • Spark Streaming是按照Batch Duration运行的,Batch Duration需要很多资源,下一次Batch Duration就不需要那么多资源了,调整资源的时候还没调整完Batch Duration运行就已经过期了。这个时候调整时间间隔。

Spark Streaming资源动态申请 

1. 在SparkContext中默认是不开启动态资源分配的,但是可以通过手动在SparkConf中配置。

// Optionally scale number of executors dynamically based on workload. Exposed for testing.val dynamicAllocationEnabled = Utils.isDynamicAllocationEnabled(_conf)if (!dynamicAllocationEnabled && //参数配置是否开启资源动态分配_conf.getBoolean("spark.dynamicAllocation.enabled", false)) {
  logWarning("Dynamic Allocation and num executors both set, thus dynamic allocation disabled.")
}

_executorAllocationManager =
  if (dynamicAllocationEnabled) {    Some(new ExecutorAllocationManager(this, listenerBus, _conf))
  } else {    None
  }
_executorAllocationManager.foreach(_.start())
  1. ExecutorAllocationManager: 有定时器会不断的去扫描Executor的情况,正在运行的Stage,要运行在不同的Executor中,要么增加Executor或者减少。

  2. ExecutorAllocationManager中schedule方法会被周期性触发进行资源动态调整。

/** * This is called at a fixed interval to regulate the number of pending executor requests * and number of executors running. * * First, adjust our requested executors based on the add time and our current needs. * Then, if the remove time for an existing executor has expired, kill the executor. * * This is factored out into its own method for testing. */private def schedule(): Unit = synchronized {  val now = clock.getTimeMillis

  updateAndSyncNumExecutorsTarget(now)

  removeTimes.retain { case (executorId, expireTime) =>
    val expired = now >= expireTime    if (expired) {
      initializing = false
      removeExecutor(executorId)
    }    !expired
  }
}
  1. 在ExecutorAllocationManager中会在线程池中定时器会不断的运行schedule.

/** * Register for scheduler callbacks to decide when to add and remove executors, and start * the scheduling task. */def start(): Unit = {
  listenerBus.addListener(listener)  val scheduleTask = new Runnable() {    override def run(): Unit = {      try {
        schedule()
      } catch {        case ct: ControlThrowable =>
          throw ct        case t: Throwable =>
          logWarning(s"Uncaught exception in thread ${Thread.currentThread().getName}", t)
      }
    }
  }// intervalMillis定时器触发时间
  executor.scheduleAtFixedRate(scheduleTask, 0, intervalMillis, TimeUnit.MILLISECONDS)
}

动态控制消费速率: Spark Streaming提供了一种弹性机制,流进来的速度和处理速度的关系,是否来得及处理数据。如果不能来得及的话,他会自动动态控制数据流进来的速度,spark.streaming.backpressure.enabled参数设置。

动态控制消费速率的原理可参考论文 Adaptive Stream Processing using Dynamic Batch Sizing


备注:

1、DT大数据梦工厂微信公众号DT_Spark 
2、IMF晚8点大数据实战YY直播频道号:68917580
3、新浪微博: http://www.weibo.com/ilovepains


本文出自 “叮咚” 博客,请务必保留此出处http://lqding.blog.51cto.com/9123978/1784901

以上是关于第17课:Spark Streaming资源动态申请和动态控制消费速率原理剖析的主要内容,如果未能解决你的问题,请参考以下文章

第6课:Spark Streaming源码解读之Job动态生成和深度思考

Spark Streaming发行版笔记17:资源动态分配和动态控制消费速率

(版本定制)第7课:Spark Streaming源码解读之JobScheduler内幕实现和深度思考

第16课:Spark Streaming源码解读之数据清理内幕彻底解密

第13课:Spark Streaming源码解读之Driver容错安全性

第12课:Spark Streaming源码解读之Executor容错安全性