private void slowerButAccurateCountTasks ( JobInfo info , JobInProgress job ) { // Count maps int totalMaps = job . numMapTasks ; int finishedMaps = 0 ; int runningMaps = 0 ; int runningMapTips = 0 ; for ( TaskInProgress tip : job . getTasks ( org . apache . hadoop . mapreduce . TaskType . MAP ) ) { if ( tip . isComplete ( ) ) { finishedMaps += 1 ; } else if ( tip . isRunning ( ) ) { runningMaps += tip . getActiveTasks ( ) . size ( ) ; runningMapTips += 1 ; } } info . totalInitedTasks = job . numMapTasks + job . numReduceTasks ; info . runningMaps = runningMaps ; infosummary . totalRunningMaps += runningMaps ; poolMgr . incRunningTasks ( info . poolName , TaskType . MAP , runningMaps ) ; info . neededSpeculativeMaps = taskSelector . neededSpeculativeMaps ( job ) ; info . neededMaps = ( totalMaps - runningMapTips - finishedMaps + info . neededSpeculativeMaps ) ; // Count reduces int totalReduces = job . numReduceTasks ; int finishedReduces = 0 ; int runningReduces = 0 ; int runningReduceTips = 0 ; for ( TaskInProgress tip : job . getTasks ( org . apache . hadoop . mapreduce . TaskType . REDUCE ) ) { if ( tip . isComplete ( ) ) { finishedReduces += 1 ; } else if ( tip . isRunning ( ) ) { runningReduces += tip . getActiveTasks ( ) . size ( ) ; runningReduceTips += 1 ; } } info . runningReduces = runningReduces ; infosummary . totalRunningReduces += runningReduces ; poolMgr . incRunningTasks ( info . poolName , TaskType . REDUCE , runningReduces ) ; if ( job . scheduleReduces ( ) ) { info . neededSpeculativeReduces = taskSelector . neededSpeculativeReduces ( job ) ; info . neededReduces = ( totalReduces - runningReduceTips - finishedReduces + info . neededSpeculativeReduces ) ; } else { info . neededReduces = 0 ; } }
Obtain task counts for a job by scanning all the tasks of all running jobs .