Hadoop之TaskTraker分析.doc

? Hadoop之TaskTraker分析 TaskTracker的工作职责之前已经和大家提过，主要负责维护，申请和监控Task，通过heartbeat和JobTracker进行通信。 ???? TaskTracker的init过程： ???? 1.读取配置文件，解析参数 ???? 2.将TaskTraker上原有的用户local files删除并新建新的dir和file ???? 3.?MapTaskAttemptID, TaskInProgress tasks = new HashMapTaskAttemptID, TaskInProgress(); 清除map ???? 4.??? this.runningTasks = new LinkedHashMapTaskAttemptID, TaskInProgress();记录task的链表??????????? this.runningJobs = new TreeMapJobID, RunningJob();记录job的id信息 ???? 5.初始化JVMManager： [java]?view plaincopyprint? mapJvmManager?=?new?JvmManagerForType(tracker.getMaxCurrentMapTasks(),??? ??????true,?tracker);?? ??reduceJvmManager?=?new?JvmManagerForType(tracker.getMaxCurrentReduceTasks(),?? ??????false,?tracker);?? ???? 6.初始化RPC，获取JobTracker client用于heartbeat通信； ???? 7.new一个 后台线程用于监听map完成的事件 [java]?view plaincopyprint? this.mapEventsFetcher?=?new?MapEventsFetcherThread();?? mapEventsFetcher.setDaemon(true);?? mapEventsFetcher.setName(?? ?????????????????????????Map-events?fetcher?for?all?reduce?tasks??+?on??+??? ?????????????????????????taskTrackerName);?? mapEventsFetcher.start();?? ??? 后台线程的run方法如下： [java]?view plaincopyprint? while?(running)?{?? ???????try?{?? ?????????List?FetchStatus?fList?=?null;?? ?????????synchronized?(runningJobs)?{?? ???????????while?(((fList?=?reducesInShuffle()).size())?==?0)?{?? ?????????????try?{?? ???????????????runningJobs.wait();?? ?????????????}?catch?(InterruptedException?e)?{?? ???????????????LOG.info(Shutting?down:??+?this.getName());?? ???????????????return;?? ?????????????}?? ???????????}?? ?????????}?? ?????????//?now?fetch?all?the?map?task?events?for?all?the?reduce?tasks?? ?????????//?possibly?belonging?to?different?jobs?? ?????????boolean?fetchAgain?=?false;?//flag?signifying?whether?we?want?to?fetch?? ?????????????????????????????????????//immediately?again.?? ?????????for?(FetchStatus?f?:?fList)?{?? ???????????long?currentTime?=?System.currentTimeMillis();?? ???????????try?{?? ?????????????//the?method?below?will?return?true?when?we?ha