yarn作业提交过程源码，yarn作业源码
yarn作业提交过程源码，yarn作业源码
记录源码细节，内部有中文注释
Client 端：
//最终通过ApplicationClientProtocol协议提交到RM端的ClientRMService内
package org.apache.hadoop.mapred;
jobclient包内
YarnRunner
public JobStatus submitJob(JobID jobId, String jobSubmitDir, Credentials ts)
throws IOException, InterruptedException {
addHistoryToken(ts);
// Construct necessary information to start the MR AM
ApplicationSubmissionContext appContext =
createApplicationSubmissionContext(conf, jobSubmitDir, ts);
// Submit to ResourceManager
try {
ApplicationId applicationId =
resMgrDelegate.submitApplication(appContext); //提交作业
ApplicationReport appMaster = resMgrDelegate
.getApplicationReport(applicationId);
ResourceMgrDelegate类
public ApplicationId
submitApplication(ApplicationSubmissionContext appContext)
throws YarnException, IOException {
return client.submitApplication(appContext);
}
public ResourceMgrDelegate(YarnConfiguration conf) {
super(ResourceMgrDelegate.class.getName());
this.conf = conf;
this.client = YarnClient.createYarnClient(); //该方法会创建YarnClientImpl,具体提交逻辑在该类里
init(conf);
start();
}
YarnClientImpl类
public ApplicationId
submitApplication(ApplicationSubmissionContext appContext)
throws YarnException, IOException {
ApplicationId applicationId = appContext.getApplicationId();
appContext.setApplicationId(applicationId);
SubmitApplicationRequest request =
Records.newRecord(SubmitApplicationRequest.class);
request.setApplicationSubmissionContext(appContext);
rmClient.submitApplication(request); //ApplicationClientProtocol rmClientRM端:
//提交只是往中央异步处理器加入RMAppEventType.START事件，异步处理，之后不等待处理结果，直接返回个简单的respone
ClientRMService内:
public SubmitApplicationResponse submitApplication(
SubmitApplicationRequest request) throws YarnException {
ApplicationSubmissionContext submissionContext = request
.getApplicationSubmissionContext();
ApplicationId applicationId = submissionContext.getApplicationId();
.....
}
}
try {
// call RMAppManager to submit application directly
rmAppManager.submitApplication(submissionContext,
System.currentTimeMillis(), false, user); //作业提交,调用的是RMAppManager中方法
LOG.info("Application with id " + applicationId.getId() +
" submitted by user " + user);
RMAuditLogger.logSuccess(user, AuditConstants.SUBMIT_APP_REQUEST,
"ClientRMService", applicationId);
} catch (YarnException e) {
LOG.info("Exception in submitting application with id " +
applicationId.getId(), e);
RMAuditLogger.logFailure(user, AuditConstants.SUBMIT_APP_REQUEST,
e.getMessage(), "ClientRMService",
"Exception in submitting application", applicationId);
throw e;
}
...
SubmitApplicationResponse response = recordFactory
.newRecordInstance(SubmitApplicationResponse.class);
return response;
protected void submitApplication(
ApplicationSubmissionContext submissionContext, long submitTime,
boolean isRecovered, String user) throws YarnException {
......
// Create RMApp
RMApp application =
new RMAppImpl(applicationId, rmContext, this.conf,
submissionContext.getApplicationName(), user,
submissionContext.getQueue(),
submissionContext, this.scheduler, this.masterService,
submitTime, submissionContext.getApplicationType());
....
}
// All done, start the RMApp
this.rmContext.getDispatcher().getEventHandler().handle(
new RMAppEvent(applicationId, isRecovered ? RMAppEventType.RECOVER:
RMAppEventType.START)); //往异步处理器增加个RMAppEvent事件,类型枚值RMAppEventType.START
//在RM内部会注册该类型的事件会用什么处理器来处理
}
在RM内部
// Register event handler for RmAppEvents
this.rmDispatcher.register(RMAppEventType.class,
new ApplicationEventDispatcher(this.rmContext));
...
//ApplicationEventDispatcher,最终会调用到RMAPPImpl来处理这个事件
public void handle(RMAppEvent event) {
this.writeLock.lock();
MAppEventType.START
try {
ApplicationId appID = event.getApplicationId();
LOG.debug("Processing event for " + appID + " of type "
+ event.getType());
final RMAppState oldState = getState();
try {
/* keep the master in sync with the state machine */
this.stateMachine.doTransition(event.getType(), event); //stateMachine通过状态工厂创建,状态工厂核心addTransition
//各种状态转变对应的处理器,有个submit应该是对应到MAppEventType.START
} catch (InvalidStateTransitonException e) {
LOG.error("Can't handle this event at current state", e);
private static final class StartAppAttemptTransition extends RMAppTransition {
public void transition(RMAppImpl app, RMAppEvent event) {
if (event.getType().equals(RMAppEventType.APP_SAVED)) {
assert app.getState().equals(RMAppState.NEW_SAVING);
RMAppStoredEvent storeEvent = (RMAppStoredEvent) event;
if(storeEvent.getStoredException() != null) {
// For HA this exception needs to be handled by giving up
// master status if we got fenced
LOG.error("Failed to store application: "
+ storeEvent.getApplicationId(),
storeEvent.getStoredException());
ExitUtil.terminate(1, storeEvent.getStoredException());
}
}
app.createNewAttempt(true); //
};
}
private void createNewAttempt(boolean startAttempt) {
ApplicationAttemptId appAttemptId =
ApplicationAttemptId.newInstance(applicationId, attempts.size() + 1);
RMAppAttempt attempt =
new RMAppAttemptImpl(appAttemptId, rmContext, scheduler, masterService,
submissionContext, conf, user); //新建个RMAppAttemptImpl
attempts.put(appAttemptId, attempt);
currentAttempt = attempt;
if(startAttempt) {
handler.handle(
new RMAppAttemptEvent(appAttemptId, RMAppAttemptEventType.START));//此处是RMAppAttemptEvent加入异步处理器的队列
//RM register可以看到其对应的处理器,最终调用的是RMAppAttemptImpl的handle方法
}
RMAppAttemptImpl类:
public void handle(RMAppAttemptEvent event) {
this.writeLock.lock();
try {
ApplicationAttemptId appAttemptID = event.getApplicationAttemptId();
LOG.debug("Processing event for " + appAttemptID + " of type "
+ event.getType());
final RMAppAttemptState oldState = getAppAttemptState();
try {
/* keep the master in sync with the state machine */
this.stateMachine.doTransition(event.getType(), event); //
} catch (InvalidStateTransitonException e) {
..
其中状态机有 .addTransition(RMAppAttemptState.NEW, RMAppAttemptState.SUBMITTED,
RMAppAttemptEventType.START, new AttemptStartedTransition())
AttemptStartedTransition的 Transition方法
...
// Add the application to the scheduler
appAttempt.eventHandler.handle(
new AppAddedSchedulerEvent(appAttempt.applicationAttemptId,
appAttempt.submissionContext.getQueue(), appAttempt.user)) //该事件即是schedulerEventType,会交给schedulerDispatcher
//该对象赋值SchedulerEventDispatcher,它在内部又维护了个类中央异步处理,run方法内都统一通过scheduler处理事件
//查看FIFO Scheduler的handle方法:
case APP_ADDED:
{
AppAddedSchedulerEvent appAddedEvent = (AppAddedSchedulerEvent) event;
addApplication(appAddedEvent.getApplicationAttemptId(), appAddedEvent
.getUser()); //
}
private synchronized void addApplication(ApplicationAttemptId appAttemptId,
String user) {
// TODO: Fix store
FiCaSchedulerApp schedulerApp =
new FiCaSchedulerApp(appAttemptId, user, DEFAULT_QUEUE, activeUsersManager,
this.rmContext);
applications.put(appAttemptId, schedulerApp);
metrics.submitApp(user, appAttemptId.getAttemptId());
LOG.info("Application Submission: " + appAttemptId.getApplicationId() +
" from " + user + ", currently active: " + applications.size());
rmContext.getDispatcher().getEventHandler().handle(
new RMAppAttemptEvent(appAttemptId,
RMAppAttemptEventType.APP_ACCEPTED)); //又是个新的状态,最终RM的ApplicationMasterLauncher与NM通信
//启动AM,AM又向RM注册,那AM实始化各个map task,reduce task是怎么做的呢
}
//该事件会ApplicationAttemptEventDispatcher来处理,在register里注册,会调用RMAppAttempImpl.handle来处理
public void handle(RMAppAttemptEvent event) {
this.writeLock.lock();
try {
ApplicationAttemptId appAttemptID = event.getApplicationAttemptId();
LOG.debug("Processing event for " + appAttemptID + " of type "
+ event.getType());
final RMAppAttemptState oldState = getAppAttemptState();
try {
/* keep the master in sync with the state machine */
this.stateMachine.doTransition(event.getType(), event); // RMAppAttemptEventType.APP_ACCEPTED会激发从什么状态到什么状态,然后执行什么事件.addTransition定义
//会到schedulered状态,再通过CONTAINER_ALLOCATED事件到ALLOCATED_SAVING状态,再通过CONTAINER_ACQURIED到
//ALLOCATED状态,再通过LAUNCHED事件到LAUNCHED状态
比如:
.addTransition(RMAppAttemptState.SCHEDULED,
RMAppAttemptState.ALLOCATED_SAVING,
RMAppAttemptEventType.CONTAINER_ALLOCATED,
new AMContainerAllocatedTransition()) //CONTAINER_ALLOCATED会激动SCHEDULED到ALLOCATED_SAVING状态,并执行CONTAINER_ALLOCATED
//最后会在nm端启动appmaster，appmaster会初始化一系列map,reduce task,再向RM注册，向RM发送heartbeat
//为task请求资源，注意心跳可能没有新的请求资源信息，再从RM内存结构里已经分配好取
//注意NM心跳到，也会执行资源分配，保留在内存结构，等appmaster来取
关键是状态机RMAPPImpl RMAppAttempImpl，内部会定义一系列的状态到状态的转换及对应的处理类
MapReduce作业提交源码分析
我们在编写MapReduce程序的时候，首先需要编写Map函数和Reduce函数。完成mapper和reducer的编写后，进行Job的配置；Job配置完成后，调用Job.submit()方法完成作业的提交。那我们思考一下，Job最终如何完成作业(job)的提交呢？粗略想一下，Job必然需要通过某种方式连接到JobTracker，因为只有这样才能将job提交到JobTracker上进行调度执行。还需要考虑一下，我们自己编写的mapper和reducer，即Jar文件如何传送到JobTracker上呢？其中有一种最简单也比较直观的方法，直接通过socket传输给JobTracker，由JobTracker再传输给TaskTracker（注意：MapReduce并没有采用这种方法）。第三个需要考虑的内容是，JobTracker如何将用户作业的配置转化成map task和reduce task。下面我们来分析一下MapReduce这些功能的实现。
首先在class Job内部通过JobClient完成作业的提交，最终由JobClient完成与JobTracker的交互功能。在JobClient的构造函数中，通过调用RPC完成与JobTracker连接的建立。
完成建立后，JobClient首先确定job相关文件的存放位置（我们上面提到mapreduce没有采用将jar即其他文件传输给JobTracker的方式，而是将这些文件保存到HDFS当中，并且可以根据用户的配置存放多份）。至于该存放目录的分配是通过调用RPC访问JobTracker的方法来进行分配的，下面看一下JobTracker的分配代码：
final Path stagingRootDir = new Path(conf.get(
"mapreduce.jobtracker.staging.root.dir",
"/tmp/Hadoop/mapred/staging"));
final FileSystem fs = stagingRootDir.getFileSystem(conf);
return fs.makeQualified(new Path(stagingRootDir, user + "/.staging")).toString();
注意上面代码所生成的stagingRootDir是所有job文件的存放目录，是一个根目录，并不单指当前job。
完成job存放目录的分配后，JobClient向JobTracker申请一个JobID（通过RPC，注意基本上JobClient与JobTracker的所有通信都是通过RPC完成的，如果下文没有显示著名也应该属于这种情况）。
JobID jobId = jobSubmitClient.getNewJobId();
下面是JobTracker.getNewJobId的具体实现：
publicsynchronized JobID getNewJobId() throws IOException {
returnnew JobID(getTrackerIdentifier(), nextJobId++);
}
获得JobID后，将该JobID与上面的stagingRootDir组合就构成了Job文件的具体存放地址的构建。进行这些相关工作后，JobClient将相关的文件存储到HDFS......余下全文>>
基于你 de 题目基于ASP的网上作业提交系统论文，
我们可以提供 1 份代码,适用于初学者 de ,
有别 de 要求也可以与我们联系,
联系我们需要提供你 de 问题和电子邮件,
有机会会帮你，肯定救急,
请用BaiduHi为我留言,
此回复针对所有来访者和需求者有效,
ES:\\C7A4B0425C8A2C6FAD8116214EE62CB9
评论暂时关闭