转载

Spark Streaming Dynamic Resource Allocation 文档

必要配置

通过下面参数开启DRA

spark.streaming.dynamicAllocation.enabled=true

设置最大最小的Executor 数目：

spark.streaming.dynamicAllocation.minExecutors=0 spark.streaming.dynamicAllocation.maxExecutors=50

可选配置

这些参数可以不用配置，都已经提供了一个较为合理的默认值

开启日志:

spark.streaming.dynamicAllocation.debug=true

设置DRA 生效延时：

spark.streaming.dynamicAllocation.delay.rounds=10

设置DRA 计算资源量时参考的周期数：

spark.streaming.dynamicAllocation.rememberBatchSize=1

设置DRA 释放资源的步调：

spark.streaming.dynamicAllocation.releaseRounds=5

设置DRA 资源额外保留比例:

spark.streaming.dynamicAllocation.reserveRate=0.2

DRA 算法说明

减少资源时，采用启发式算法。根据之前周期的处理时间，计算需要保留的资源量(A)，然后尝试分多轮试探性的减少(B),每个计算周期都会重复A,B动作，最后会收敛到一个具体的数值。

如果一旦发生延时，则会立马向Yarn申请 spark.streaming.dynamicAllocation.maxExecutors 个Executor，以保证可以最快速度消除延时。富余出来的资源会通过减少资源的动作慢慢进行减少，让程序趋于稳定。

发生减少资源的动作，则剔除的掉的Executor 会被立刻(几毫秒/纳秒)屏蔽，并且不再分配Task,之后再由Yarn异步移除。

添加资源的动作，则由Yarn决定

注意事项

请务必保证你Package 的App包不包含spark 相关的组件。否则你会看到自己的设置并不生效，因为运行的时候用了你的App里的spark-core,spark-streaming jar包了。

一些可以参考的调整

如果系统趋向稳定后，经过人工观察发现其实还可以再降资源，则可以尝试调低

spark.streaming.dynamicAllocation.releaseRounds=5 spark.streaming.dynamicAllocation.reserveRate=0.2

建议releaseRounds 不低于2，reserveRate 不低于0.05。避免系统发生颠簸。

测试代码

object IamGod {   def main(args: Array[String]): Unit = {      def createContext = {       val conf = new SparkConf().setAppName("DRA Test")       val ssc = new StreamingContext(conf, Seconds(30))        val items1 = Seq.fill(30)(Seq((10 + scala.util.Random.nextInt(10)) * 1000))       val items2 = Seq.fill(30)(Seq((30 + scala.util.Random.nextInt(10)) * 1000))       val items3 = Seq.fill(30)(Seq((20 + scala.util.Random.nextInt(10)) * 1000))        val fileInput = new TestInputStream[Int](ssc, items1 ++ items2 ++ items3, 10)        val logs = fileInput.map(f => Thread.sleep(f))        logs.foreachRDD { rdd =>         rdd.count()       }        ssc     }      val ssc = createContext      ssc.start()     ssc.awaitTermination()    }  }

前面引用了一个测试类：

class TestInputStream[T: ClassTag](_ssc: StreamingContext, input: Seq[Seq[T]], numPartitions: Int)   extends InputDStream[T](_ssc) {    def start() {}    def stop() {}    def compute(validTime: Time): Option[RDD[T]] = {     logInfo("Computing RDD for time " + validTime)     val index = ((validTime - zeroTime) / slideDuration - 1).toInt     val selectedInput = if (index < input.size) input(index) else Seq[T]()      // lets us test cases where RDDs are not created     if (selectedInput == null) {       return None     }      // Report the input data's information to InputInfoTracker for testing     val inputInfo = StreamInputInfo(id, selectedInput.length.toLong)     ssc.scheduler.inputInfoTracker.reportInfo(validTime, inputInfo)      val rdd = ssc.sc.makeRDD(selectedInput, numPartitions)     logInfo("Created RDD " + rdd.id + " with " + selectedInput)     Some(rdd)   } }

原文 http://www.jianshu.com/p/9c6cdf429b52

正文到此结束