背景
顺着昨天spark standalone实现那篇文章继续扯淡,看看Mesos Scheduler的两种实现的异同。
对我来说,回过头再仔细看Spark在这一层的实现,思路又清晰了许多。
Mesos粗粒度
CoarseMesosSchedulerBackend,是mesos的粗粒度scheduler backend实现。
简单说一下mesos的Scheduler,提供的回调函数,及spark实现的逻辑:
Mesos Scheduler接口 | 触发场景 | spark实现逻辑 |
void registered( SchedulerDriver driver, FrameworkID frameworkId, MasterInfo masterInfo); |
当Scheduler成向mesos master注册之后回调, 会返回唯一的framework id |
得到framework的id,作为appId |
void reregistered( SchedulerDriver driver, MasterInfo masterInfo) |
是mesos换了个新选举出来的master的时候触发, 前提是该scheduler之前已经注册过了 |
没有实现。 |
void resourceOffers( SchedulerDriver driver, List<Offer> offers) |
mesos提供了一批可用的资源,让scheduler可以使用或拒绝这些资源。 每个Offer是以slave为单位的,即以slave为单位的资源列表。 |
得到mesos的Offers列表,只要已经启动的executor还不足够, 那么从资源列表里继续获取资源,启动CoarseGrainedExecutorBackend。 |
void offerRescinded( SchedulerDriver driver, OfferID offerId) |
当请求的offer不可用时回调(可能是slave lost了之类的原因导致的), 如果在这上面继续起task的话会报Task Lost的状态。 |
没有实现。spark在task scheduler层面对lost的task有自己的处理,。 |
void statusUpdate( SchedulerDriver driver, TaskStatus status) |
task状态更新回调,可能是finish了,可能是lost了,可能是fail了等等 | 得到finished、failed、lost等task状态,更新内存里维护的task状态, 并触发新一轮的reviveOffers,即通过task scheduler继续把resource分配给需要的task并执行它们。 |
void frameworkMessage( SchedulerDriver driver, ExecutorID executorId, SlaveID slaveId, byte[] data) |
用于接收executor主动发的消息 | 没有实现。mesos虽然在内部提供了这种msg接口,貌似不是很稳定。 使用方可以使用自己的RPC来做executor与scheduler的通信,如果真的需要的话。 |
void disconnected( SchedulerDriver driver) |
当scheduler与master断开的时候触发, 原因可能是master挂了,或者master换了等等。 |
没有实现。spark前面就没有实现master新选举的接口。 |
void slaveLost( SchedulerDriver driver, SlaveID slaveId) |
通知某个slave lost了,以便framework进行任务的rescheduler或其他逻辑 | spark把slave lost和executor lost一起处理了。 处理逻辑就是执行RemoveExecutor操作,最终调用TaskScheduler的executorLost方法,把executor的状态移除, 并且会继续向上调用DAGScheduler的handleExecutorLost方法。 因executor lost可能会影响到shuffle数据,这部分还需要BlockManager感知。 |
void executorLost( SchedulerDriver driver, ExecutorID executorId, SlaveID slaveId, int status) |
通知某个slave上的executor挂了。 | 同上 |
void error( SchedulerDriver driver, String message) |
scheduler或scheduler driver发送错误触发 | 没有实现 |
另一方面,要说明一下mesos的SchedulerDriver。
它相当于是Scheduler与mesos通信的一个连接人,一方面是管理Scheduler的生命周期,二方面是与mesos交互,让它启停task。
在初始化SchedulerDriver的时候,向里面传入spark自己实现的Scheduler就可以了,即CoarseMesosSchedulerBackend或MesosSchedulerBackend。在每个mesos的Scheduler接口的回调方法里,都会传回SchedulerDriver,以对应可以进行scheduler的启停和task的启停管理。
CoarseMesosSchedulerBackend内部主要维护的信息为:slave与executor的对应关系,executor与task的对应关系,task与slave的对应关系。
Mesos细粒度
翻译下注释:
细粒度模式下,允许多个app共享nodes,包括空间,即不同app的tasks可以跑同几个core,和时间,即一个core可以切换ownership。
这块共享的控制,在mesos端,所以spark在实现的时候,其实差别和难度都不大。
MesosSchedulerBackend的实现:
在resourceOffers逻辑里,获得mesos提供的slave资源后,直接在里面调用scheduler的resourceOffers,即TaskSchedulerImpl的分配task的逻辑。也就是说,会按优先级,从active task sets(来自多个app)里选择并直接launch task。而粗粒度里的做法,是先启动executorbackend,把资源准备好,进程先拉起,供app去launch task。
其他回调接口的逻辑是大同小异的。
还有一点不同之处,粗粒度模式下executor的实现使用的是与standalone模式相同的CoarseGrainedExecutorBackend。在细粒度模式下,executor的实现是MesosExecutorBackend,实现了spark的ExecutorBackend和mesos的MesosExecutor。实际上,在类里面,还是使用的spark的executor,在对应的回调里执行start/kill task这样的操作。另外,mesos的ExecutorDriver用于负责与mesos通信,比如传递一些状态更新的消息,或有特殊的msg要发送。executor这块的差别无关紧要。
在我看来,executor这块最终一定是落到了spark的executor上,里面有一个线程池,可以跑spark的ShuffleMapTask或ResultTask。而mesos粗、细粒度的Scheduler实现,最大区别在于resourceOffers的逻辑,是怎么处理分配的进程资源:粗粒度是预先拉起执行进程,而细粒度是直接通过TaskScheduler来摆放执行线程了。
粗细粒度分别适合跑什么样的任务,可以具体见 官方文档这一节
全文完 :)