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.2DRA 算法说明
减少资源时，采用启发式算法。根据之前周期的处理时间，计算需要保留的资源量(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)
}
}
