在上文《Spark技术内幕:Stage划分及提交源代码分析》中,我们分析了Stage的生成和提交。可是Stage的提交,仅仅是DAGScheduler完毕了对DAG的划分,生成了一个计算拓扑,即须要依照顺序计算的Stage,Stage中包括了能够以partition为单位并行计算的Task。我们并没有分析Stage中得Task是怎样生成而且终于提交到Executor中去的。
这就是本文的主题。
从org.apache.spark.scheduler.DAGScheduler#submitMissingTasks開始,分析Stage是怎样生成TaskSet的。
假设一个Stage的全部的parent stage都已经计算完毕或者存在于cache中。那么他会调用submitMissingTasks来提交该Stage所包括的Tasks。
org.apache.spark.scheduler.DAGScheduler#submitMissingTasks的计算流程例如以下:
- 首先得到RDD中须要计算的partition,对于Shuffle类型的stage。须要推断stage中是否缓存了该结果;对于Result类型的Final Stage。则推断计算Job中该partition是否已经计算完毕。
- 序列化task的binary。Executor能够通过广播变量得到它。每一个task执行的时候首先会反序列化。这样在不同的executor上执行的task是隔离的,不会相互影响。
- 为每一个须要计算的partition生成一个task:对于Shuffle类型依赖的Stage,生成ShuffleMapTask类型的task;对于Result类型的Stage,生成一个ResultTask类型的task
- 确保Task是能够被序列化的。由于不同的cluster有不同的taskScheduler,在这里推断能够简化逻辑。保证TaskSet的task都是能够序列化的
- 通过TaskScheduler提交TaskSet。
pipeline。能够称为大数据处理的基石。仅仅有数据进行pipeline处理,才干将其放到集群中去执行。
对于一个task来说,它从数据源获得逻辑。然后依照拓扑顺序,顺序执行(实际上是调用rdd的compute)。
private[spark] class TaskSet( val tasks: Array[Task[_]], val stageId: Int, val attempt: Int, val priority: Int, val properties: Properties) { val id: String = stageId + "." + attempt override def toString: String = "TaskSet " + id }
- org.apache.spark.scheduler.TaskSchedulerImpl#submitTasks
- org.apache.spark.scheduler.SchedulableBuilder#addTaskSetManager
- org.apache.spark.scheduler.cluster.CoarseGrainedSchedulerBackend#reviveOffers
- org.apache.spark.scheduler.cluster.CoarseGrainedSchedulerBackend.DriverActor#makeOffers
- org.apache.spark.scheduler.TaskSchedulerImpl#resourceOffers
- org.apache.spark.scheduler.cluster.CoarseGrainedSchedulerBackend.DriverActor#launchTasks
- org.apache.spark.executor.CoarseGrainedExecutorBackend.receiveWithLogging#launchTask
- org.apache.spark.executor.Executor#launchTask
def launchTask( context: ExecutorBackend, taskId: Long, taskName: String, serializedTask: ByteBuffer) { val tr = new TaskRunner(context, taskId, taskName, serializedTask) runningTasks.put(taskId, tr) threadPool.execute(tr) // 開始在executor中执行 }
final def run(attemptId: Long): T = { context = new TaskContext(stageId, partitionId, attemptId, runningLocally = false) context.taskMetrics.hostname = Utils.localHostName() taskThread = Thread.currentThread() if (_killed) { kill(interruptThread = false) } runTask(context) }
对于原来提到的两种Task,即
- org.apache.spark.scheduler.ShuffleMapTask
- org.apache.spark.scheduler.ResultTask
override def runTask(context: TaskContext): U = { // Deserialize the RDD and the func using the broadcast variables. val ser = SparkEnv.get.closureSerializer.newInstance() val (rdd, func) = ser.deserialize[(RDD[T], (TaskContext, Iterator[T]) => U)]( ByteBuffer.wrap(taskBinary.value), Thread.currentThread.getContextClassLoader) metrics = Some(context.taskMetrics) try { func(context, rdd.iterator(partition, context)) } finally { context.markTaskCompleted() } }
override def runTask(context: TaskContext): MapStatus = { // Deserialize the RDD using the broadcast variable. val ser = SparkEnv.get.closureSerializer.newInstance() val (rdd, dep) = ser.deserialize[(RDD[_], ShuffleDependency[_, _, _])]( ByteBuffer.wrap(taskBinary.value), Thread.currentThread.getContextClassLoader) //此处的taskBinary即为在org.apache.spark.scheduler.DAGScheduler#submitMissingTasks序列化的task的广播变量取得的 metrics = Some(context.taskMetrics) var writer: ShuffleWriter[Any, Any] = null try { val manager = SparkEnv.get.shuffleManager writer = manager.getWriter[Any, Any](dep.shuffleHandle, partitionId, context) writer.write(rdd.iterator(partition, context).asInstanceOf[Iterator[_ <: Product2[Any, Any]]]) // 将rdd计算的结果写入memory或者disk return writer.stop(success = true).get } catch { case e: Exception => if (writer != null) { writer.stop(success = false) } throw e } finally { context.markTaskCompleted() } }
对于这两个task都用到的taskBinary,即为在org.apache.spark.scheduler.DAGScheduler#submitMissingTasks序列化的task的广播变量取得的。
以后的每一个周末。都会奉上某个细节的实现。