参考Blog
DAGScheduler.submitMissingTasks(stage: Stage, jobId: Int)
方法中的最后几行，有一个
taskScheduler.submitTasks(new TaskSet(
tasks.toArray, stage.id, stage.latestInfo.attemptId, jobId, properties))
在 Standalone 模式下 taskScheduler 的实现类是 TaskSchedulerImpl
所以查看 TaskSchedulerImpl.submitTasks(TaskSet)方法
这里的 backend 的实现类是 SparkDeploySchedulerBackend ，然而 SparkDeploySchedulerBackend 没有 reviveOffers() 方法。 SparkDeploySchedulerBackend extends CoarseGrainedSchedulerBackend， 这里面是有 reviveOffers() 方法的。
这里的 driverEndpoint 是在 CoarseGrainedSchedulerBackend.start() 方法初始化的。
这个 rpcEnv 默认的实现类是 NettyRpcEnv。
为何而知？ SparkContext 成员变量赋值时
OK ,继续查看 driverEndpoint = rpcEnv.setupEndpoint(ENDPOINT_NAME, createDriverEndpoint(properties))
ENDPOINT_NAME 变量值和 createDriverEndpoint 方法如下
|
1 2 3 4 5 6 7 |
val ENDPOINT_NAME = "CoarseGrainedScheduler" createDriverEndpoint(properties) ... protected def createDriverEndpoint(properties: Seq[(String, String)]): DriverEndpoint = { new DriverEndpoint(rpcEnv, properties) } |
DriverEndpoint extends ThreadSafeRpcEndpoint
而 ThreadSafeRpcEndpoint extends RpcEndpoint
RpcEndpoint 的备注如下
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 |
/** * An end point for the RPC that defines what functions to trigger given a message. * * It is guaranteed that `onStart`, `receive` and `onStop` will be called in sequence. * * The life-cycle of an endpoint is: * * constructor -> onStart -> receive* -> onStop * * Note: `receive` can be called concurrently. If you want `receive` to be thread-safe, please use * [[ThreadSafeRpcEndpoint]] * * If any error is thrown from one of [[RpcEndpoint]] methods except `onError`, `onError` will be * invoked with the cause. If `onError` throws an error, [[RpcEnv]] will ignore it. */ |
有价值的是constructor -> onStart -> receive* -> onStop
OK ,继续查看 driverEndpoint = rpcEnv.setupEndpoint(ENDPOINT_NAME, createDriverEndpoint(properties)) 的 NettyRpcEnv.setupEndpoint 方法
这里返回的RpcEndpointRef为NettyRpcEndpointRef类型，而RpcEndpointRef则是一个远程RpcEndpoint的引用，通过它可以给远程RpcEndpoint发送消息，可以是同步可以是异步，它映射一个地址。这么看来，我们在远端（ps：另外的机器或者进程）注册了一个RpcEndpoint，即DriverEndpoint，而在本地端（当前机器或者进程）则持有一个RpcEndpoint的引用，即NettyRpcEndpointRef，可以由它来往远端发送消息，那么发送的是什么消息呢？我们现在返回CoarseGrainedSchedulerBackend中的reviveOffers()方法，发现发送的是ReviveOffers消息。这里只是发送，具体处理还要看远端的RpcEndpoint，即DriverEndpoint。通过上面我们可以知道，RpcEndpoint的服务流程为onStart()–>receive()–> onStop()，每当消息来临时，DriverEndpoint都会调用receive()方法来处理。关键代码如下：
原创文章，转载请注明： 转载自LoserZhao – 诗和远方[ http://www.loserzhao.com/ ]
本文链接地址: http://www.loserzhao.com/bigdata/spark-task-schedule-source-analysis.html
文章的脚注信息由WordPress的wp-posturl插件自动生成
0 条评论。