大数据IMF传奇行动绝密课程第34课：Stage划分和Task最佳位置算法解密

Stage划分和Task最佳位置算法解密

一、Job Stage划分算法解密

二、Task最佳位置算法实现解密

一：Stage划分算法解密

1、Spark Application中可以因为不同的Action出发众多的Job，也就是说一个Application中可以有很多的Job，每个Job是由一个或者多个Stage构成的，后面的Stage依赖于前面的Stage，也就是说只有前面依赖的Stage计算完毕后，后面的Stage才会运行；

2、Stage划分的依据就是宽依赖，产生宽依赖的情况例如：reduceByKey、groupByKey等

3、由Action（例如collect）导致了SparkContext.runJob的执行，最终导致了DAGScheduler中的submitJob的执行，其核心是通过发送一个case class JobSubmitted对象给eventProcessLoop，其中JobSubmitted源码如下：

private[scheduler] case class JobSubmitted(
jobId: Int,
finalRDD: RDD[_],
func: (TaskContext, Iterator[_]) => _,
partitions: Array[Int],
callSite: CallSite,
listener: JobListener,
properties: Properties = null)
extends DAGSchedulerEvent

eventProcessLoop是DAGSchedulerEventProcessLoop的具体实例，而DAGSchedulerEventProcessLoop是EventLoop的子类，具体实现EventLoop的onReceive方法，onReceive方法转过来回调doOnReceive

4、在doOnReceive中通过模式匹配的方式把执行路由到 case JobSubmitted(jobId, rdd, func, partitions, callSite, listener, properties) =>

dagScheduler.handleJobSubmitted(jobId, rdd, func, partitions, callSite, listener, properties)

开辟了线程，线程会不断的循环消息队列，post的时候会把消息放到队列中，线程从eventQueue中获得消息（LinkedBlockingQueue类型，可以给他发信息），DAGSchedulerEventProcessLoop中doOnReceive处理收到的消息

为何要单独弄消息循环器：1、异步处理多Job，不用等待，非阻塞；2、便于扩展，可以处理所有的消息

5、在handleJobSubmitted中，首先创建finalStage，会有异常（当HDFS文件被修改），创建finalStage时会建立父Stage的依赖链条

补充说明：所谓的missing就是说要进行当前的计算了

二、Task任务本地性算法实现

1、在submitMissingTasks中会通过调用以下代码来获得任务的本地性

val taskIdToLocations: Map[Int, Seq[TaskLocation]] = try {
stage match {
case s: ShuffleMapStage =>
partitionsToCompute.map { id => (id, getPreferredLocs(stage.rdd, id))}.toMap
case s: ResultStage =>
val job = s.activeJob.get
partitionsToCompute.map { id =>
val p = s.partitions(id)
(id, getPreferredLocs(stage.rdd, p))
}.toMap
}
}

2、具体一个Partition中数据本地性的算法实现为下述代码

private[spark]
def getPreferredLocs(rdd: RDD[_], partition: Int): Seq[TaskLocation] = {
getPreferredLocsInternal(rdd, partition, new HashSet)
}

在具体算法实现的时候，首先查询DAGScheduler的内存数据结构中是否存在当前Partition的数据本地性的信息，如果有的话则直接返回；如果没有首先会调用rdd.getPreferedLocations

例如想让Spark运行在HBase上或者一种现在还没有直接支持的数据库上面，此时开发者需要自定义RDD，为了保证Task计算的数据本地性，最为关键的方式就是必须实现RDD的getPreferredLocations

3、DAGScheduler计算数据本地性的时候巧妙的借助了RDD自身的getPreferredLocations中的数据，最大化优化了效率，因为getPreferredLocations中表明了每个Partition的数据本地性，虽然当前Partition可能被persist或者checkpoint，但是persist或者checkpoiint默认情况下肯定是和getPreferredLocations中的Partition的数据本地性是一致的，所以这就极大的简化了Task数据本地性算法的实现效率的优化