public boolean monitorAndPrintJob ( JobConf conf , RunningJob job ) throws IOException , InterruptedException { String lastReport = null ; TaskStatusFilter filter ; filter = getTaskOutputFilter ( conf ) ; JobID jobId = job . getID ( ) ; LOG . info ( "Running job: " + jobId ) ; int eventCounter = 0 ; boolean profiling = conf . getProfileEnabled ( ) ; Configuration . IntegerRanges mapRanges = conf . getProfileTaskRange ( true ) ; Configuration . IntegerRanges reduceRanges = conf . getProfileTaskRange ( false ) ; while ( ! job . isComplete ( ) ) { Thread . sleep ( MAX_JOBPROFILE_AGE ) ; String report = ( " map " + StringUtils . formatPercent ( job . mapProgress ( ) , 0 ) + " reduce " + StringUtils . formatPercent ( job . reduceProgress ( ) , 0 ) ) ; if ( ! report . equals ( lastReport ) ) { LOG . info ( report ) ; lastReport = report ; } TaskCompletionEvent [ ] events = job . getTaskCompletionEvents ( eventCounter ) ; eventCounter += events . length ; for ( TaskCompletionEvent event : events ) { TaskCompletionEvent . Status status = event . getTaskStatus ( ) ; if ( profiling && ( status == TaskCompletionEvent . Status . SUCCEEDED || status == TaskCompletionEvent . Status . FAILED ) && ( event . isMap ? mapRanges : reduceRanges ) . isIncluded ( event . idWithinJob ( ) ) ) { downloadProfile ( event ) ; } switch ( filter ) { case NONE : break ; case SUCCEEDED : if ( event . getTaskStatus ( ) == TaskCompletionEvent . Status . SUCCEEDED ) { LOG . info ( event . toString ( ) ) ; displayTaskLogs ( event . getTaskAttemptId ( ) , event . getTaskTrackerHttp ( ) ) ; } break ; case FAILED : if ( event . getTaskStatus ( ) == TaskCompletionEvent . Status . FAILED ) { LOG . info ( event . toString ( ) ) ; // Displaying the task diagnostic information TaskAttemptID taskId = event . getTaskAttemptId ( ) ; String [ ] taskDiagnostics = jobSubmitClient . getTaskDiagnostics ( taskId ) ; if ( taskDiagnostics != null ) { for ( String diagnostics : taskDiagnostics ) { System . err . println ( diagnostics ) ; } } // Displaying the task logs displayTaskLogs ( event . getTaskAttemptId ( ) , event . getTaskTrackerHttp ( ) ) ; } break ; case KILLED : if ( event . getTaskStatus ( ) == TaskCompletionEvent . Status . KILLED ) { LOG . info ( event . toString ( ) ) ; } break ; case ALL : LOG . info ( event . toString ( ) ) ; displayTaskLogs ( event . getTaskAttemptId ( ) , event . getTaskTrackerHttp ( ) ) ; break ; } } } LOG . info ( "Job complete: " + jobId ) ; Counters counters = job . getCounters ( ) ; if ( counters != null ) { counters . log ( LOG ) ; } return job . isSuccessful ( ) ; }
Monitor a job and print status in real - time as progress is made and tasks fail .