Spark技术内幕: Task向Executor提交的源码解析

标签：des   style   http   color   io   os   ar   java   for   

从org.apache.spark.scheduler.DAGScheduler#submitMissingTasks开始，分析Stage是如何生成TaskSet的。

如果一个Stage的所有的parent stage都已经计算完成或者存在于cache中，那么他会调用submitMissingTasks来提交该Stage所包含的Tasks。

org.apache.spark.scheduler.DAGScheduler#submitMissingTasks的计算流程如下：

1. 首先得到RDD中需要计算的partition，对于Shuffle类型的stage，需要判断stage中是否缓存了该结果；对于Result类型的Final Stage，则判断计算Job中该partition是否已经计算完成。
2. 序列化task的binary。Executor可以通过广播变量得到它。每个task运行的时候首先会反序列化。这样在不同的executor上运行的task是隔离的，不会相互影响。
3. 为每个需要计算的partition生成一个task：对于Shuffle类型依赖的Stage，生成ShuffleMapTask类型的task；对于Result类型的Stage，生成一个ResultTask类型的task
4. 确保Task是可以被序列化的。因为不同的cluster有不同的taskScheduler，在这里判断可以简化逻辑；保证TaskSet的task都是可以序列化的
5. 通过TaskScheduler提交TaskSet。

[java] view plaincopy

1. private[spark] class TaskSet(  
2.     val tasks: Array[Task[_]],  
3.     val stageId: Int,  
4.     val attempt: Int,  
5.     val priority: Int,  
6.     val properties: Properties) {  
7.     val id: String = stageId + "." + attempt  
8.   
9.   override def toString: String = "TaskSet " + id  
10. }  

[java] view plaincopy

1. def launchTask(  
2.     context: ExecutorBackend, taskId: Long, taskName: String, serializedTask: ByteBuffer) {  
3.   val tr = new TaskRunner(context, taskId, taskName, serializedTask)  
4.   runningTasks.put(taskId, tr)  
5.   threadPool.execute(tr) // 开始在executor中运行  
6. }  

[java] view plaincopy

1. final def run(attemptId: Long): T = {  
2.    context = new TaskContext(stageId, partitionId, attemptId, runningLocally = false)  
3.    context.taskMetrics.hostname = Utils.localHostName()  
4.    taskThread = Thread.currentThread()  
5.    if (_killed) {  
6.      kill(interruptThread = false)  
7.    }  
8.    runTask(context)  
9.  }  

对于原来提到的两种Task，即

1.  org.apache.spark.scheduler.ShuffleMapTask
2.  org.apache.spark.scheduler.ResultTask

[java] view plaincopy

1. override def runTask(context: TaskContext): U = {  
2.   // Deserialize the RDD and the func using the broadcast variables.  
3.   val ser = SparkEnv.get.closureSerializer.newInstance()  
4.   val (rdd, func) = ser.deserialize[(RDD[T], (TaskContext, Iterator[T]) => U)](  
5.     ByteBuffer.wrap(taskBinary.value), Thread.currentThread.getContextClassLoader)  
6.   
7.   metrics = Some(context.taskMetrics)  
8.   try {  
9.     func(context, rdd.iterator(partition, context))  
10.   } finally {  
11.     context.markTaskCompleted()  
12.   }  
13. }  

[java] view plaincopy

1. override def runTask(context: TaskContext): MapStatus = {  
2.   // Deserialize the RDD using the broadcast variable.  
3.   val ser = SparkEnv.get.closureSerializer.newInstance()  
4.   val (rdd, dep) = ser.deserialize[(RDD[_], ShuffleDependency[_, _, _])](  
5.     ByteBuffer.wrap(taskBinary.value), Thread.currentThread.getContextClassLoader)  
6.     //此处的taskBinary即为在org.apache.spark.scheduler.DAGScheduler#submitMissingTasks序列化的task的广播变量取得的  
7.   
8.   metrics = Some(context.taskMetrics)  
9.   var writer: ShuffleWriter[Any, Any] = null  
10.   try {  
11.     val manager = SparkEnv.get.shuffleManager  
12.     writer = manager.getWriter[Any, Any](dep.shuffleHandle, partitionId, context)  
13.     writer.write(rdd.iterator(partition, context).asInstanceOf[Iterator[_ <: Product2[Any, Any]]]) // 将rdd计算的结果写入memory或者disk  
14.     return writer.stop(success = true).get  
15.   } catch {  
16.     case e: Exception =>  
17.       if (writer != null) {  
18.         writer.stop(success = false)  
19.       }  
20.       throw e  
21.   } finally {  
22.     context.markTaskCompleted()  
23.   }  

Spark技术内幕: Task向Executor提交的源码解析

标签：des   style   http   color   io   os   ar   java   for   

原文地址：http://www.cnblogs.com/yido9932/p/4053371.html