第一篇：Hadoop之JobTrack分析

Hadoop之JobTrack分析

1.client端指定Job的各種參數配置之后調用job.waitForCompletion(true)方法提交Job給JobTracker，等待Job 完成。

[java] view plaincopyprint?

1.public void submit()throws IOException, InterruptedException, 2.ClassNotFoundException { 3.ensureState(JobState.DEFINE);//檢查JobState狀態

4.setUseNewAPI();//檢查及設置是否使用新的MapReduce API

5.6.// Connect to the JobTracker and submit the job

7.connect();//鏈接JobTracker

8.info = jobClient.submitJobInternal(conf);//將job信息提交

9.super.setJobID(info.getID());

10.state = JobState.RUNNING;//更改job狀態

11.}

以上代碼主要有兩步驟,連接JobTracker并提交Job信息。connect方法主要是實例化JobClient對象，包括設置JobConf和init工作：

[java] view plaincopyprint?

1.public void init(JobConf conf)throws IOException {

2.String tracker = conf.get(“mapred.job.tracker”, “local”);//讀取配置文件信息用于判斷該Job是運行于本地單機模式還是分布式模式

3.tasklogtimeout = conf.getInt（4.TASKLOG_PULL_TIMEOUT_KEY, DEFAULT_TASKLOG_TIMEOUT);5.this.ugi = UserGroupInformation.getCurrentUser();

6.if(“local”.equals(tracker)){//如果是單機模式，new LocalJobRunner

7.conf.setNumMapTasks(1);

8.this.jobSubmitClient = new LocalJobRunner(conf);9.} else {

10.this.jobSubmitClient = createRPCProxy(JobTracker.getAddress(conf), conf);

11.} 12.}

分布式模式下就會創建一個RPC代理鏈接：

[java] view plaincopyprint?

1.public static VersionedProtocol getProxy(2.Class protocol，3.long clientVersion, InetSocketAddress addr, UserGroupInformation ticket，4.Configuration conf, SocketFactory factory, int rpcTimeout)throws IOException { 5.6.if(UserGroupInformation.isSecurityEnabled()){ 7.SaslRpcServer.init(conf);8.}

9.VersionedProtocol proxy =

10.(VersionedProtocol)Proxy.newProxyInstance（11.protocol.getClassLoader(), new Class[] { protocol }，12.new Invoker(protocol, addr, ticket, conf, factory, rpcTimeout));

13.long serverVersion = proxy.getProtocolVersion(protocol.getName(), 14.clientVersion);15.if(serverVersion == clientVersion){ 16.return proxy;17.} else {

18.throw new VersionMismatch(protocol.getName(), clientVersion, 19.serverVersion);20.} 21.}

從上述代碼可以看出hadoop實際上使用了Java自帶的Proxy API來實現Remote Procedure Call 初始完之后，需要提交job [java] view plaincopyprint?

1.info = jobClient.submitJobInternal(conf);//將job信息提交

submit方法做以下幾件事情：

1.將conf中目錄名字替換成hdfs代理的名字

2.檢查output是否合法：比如路徑是否已經存在，是否是明確的3.將數據分成多個split并放到hdfs上面，寫入job.xml文件

4.調用JobTracker的submitJob方法

該方法主要新建JobInProgress對象，然后檢查訪問權限和系統參數是否滿足job，最后addJob：

[java] view plaincopyprint?

1.private synchronized JobStatus addJob(JobID jobId, JobInProgress job)2.throws IOException { 3.totalSubmissions++;4.5.synchronized(jobs){

6.synchronized(taskScheduler){

7.jobs.put(job.getProfile().getJobID(), job);

8.for(JobInProgressListener listener : jobInProgressListeners){ 9.listener.jobAdded(job);10.} 11.} 12.}

13.myInstrumentation.submitJob(job.getJobConf(), jobId);14.job.getQueueMetrics().submitJob(job.getJobConf(), jobId);15.16.LOG.info(“Job ” + jobId + “ added successfully for user '”

17.+ job.getJobConf().getUser()+ “' to queue '”

18.+ job.getJobConf().getQueueName()+ “'”);19.AuditLogger.logSuccess(job.getUser()，20.Operation.SUBMIT_JOB.name(), jobId.toString());21.return job.getStatus();22.}

totalSubmissions記錄client端提交job到JobTracker的次數。而jobs則是JobTracker所有可以管理的job的映射表

Map jobs = Collections.synchronizedMap(new TreeMap());taskScheduler是用于調度job先后執行策略的，其類圖如下所示：

hadoop job調度機制； public enum SchedulingMode { FAIR, FIFO } 1.公平調度FairScheduler 對于每個用戶而言，分布式資源是公平分配的，每個用戶都有一個job池，假若某個用戶目前所占有的資源很多，對于其他用戶而言是不公平的，那么調度器就會殺掉占有資源多的用戶的一些task，釋放資源供他人使用 2.容量調度JobQueueTaskScheduler 在分布式系統上維護多個隊列，每個隊列都有一定的容量，每個隊列中的job按照FIFO的策略進行調度。隊列中可以包含隊列。

兩個Scheduler都要實現TaskScheduler的public synchronized List assignTasks(TaskTracker tracker)方法，該方法通過具體的計算生成可以分配的task

接下來看看JobTracker的工作： 記錄更新JobTracker重試的次數：

[java] view plaincopyprint?

1.while(true){ 2.try {

3.recoveryManager.updateRestartCount();4.break;

5.} catch(IOException ioe){

6.LOG.warn(“Failed to initialize recovery manager.”, ioe);7.// wait for some time

8.Thread.sleep(FS_ACCESS_RETRY_PERIOD);9.LOG.warn(“Retrying...”);10.} 11.}

啟動Job調度器,默認是FairScheduler: taskScheduler.start();主要是初始化一些管理對象，比如job pool管理池

[java] view plaincopyprint?

1.// Initialize other pieces of the scheduler

2.jobInitializer = new JobInitializer(conf, taskTrackerManager);3.taskTrackerManager.addJobInProgressListener(jobListener);4.poolMgr = new PoolManager(this);5.poolMgr.initialize();

6.loadMgr =(LoadManager)ReflectionUtils.newInstance(7.conf.getClass(“mapred.fairscheduler.loadmanager”, 8.CapBasedLoadManager.class, LoadManager.class), conf);9.loadMgr.setTaskTrackerManager(taskTrackerManager);10.loadMgr.setEventLog(eventLog);11.loadMgr.start();

12.taskSelector =(TaskSelector)ReflectionUtils.newInstance(13.conf.getClass(“mapred.fairscheduler.taskselector”, 14.DefaultTaskSelector.class, TaskSelector.class), conf);15.taskSelector.setTaskTrackerManager(taskTrackerManager);16.taskSelector.start();

[java] view plaincopyprint?

1.JobInitializer有一個確定大小的ExecutorService threadPool，每個thread用于初始化job

[java] view plaincopyprint?

1.try {

2.JobStatus prevStatus =(JobStatus)job.getStatus().clone();3.LOG.info(“Initializing ” + job.getJobID());4.job.initTasks();

5.// Inform the listeners if the job state has changed 6.// Note : that the job will be in PREP state.7.JobStatus newStatus =(JobStatus)job.getStatus().clone();8.if(prevStatus.getRunState()!= newStatus.getRunState()){ 9.JobStatusChangeEvent event =

10.new JobStatusChangeEvent(job, EventType.RUN_STATE_CHANGED, prevStatus，11.newStatus);

12.synchronized(JobTracker.this){ 13.updateJobInProgressListeners(event);14.} 15.} 16.}

初始化操作主要用于初始化生成tasks然后通知其他的監聽者執行其他操作。initTasks主要處理以下工作：

[java] view plaincopyprint?

1.// 記錄用戶提交的運行的job信息

2.try {

3.userUGI.doAs(new PrivilegedExceptionAction