private void moveTaskOutputs ( TaskAttemptContext context , FileSystem fs , Path jobOutputDir , Path taskOutput ) throws IOException { TaskAttemptID attemptId = context . getTaskAttemptID ( ) ; context . progress ( ) ; if ( fs . isFile ( taskOutput ) ) { Path finalOutputPath = getFinalPath ( jobOutputDir , taskOutput , workPath ) ; if ( ! fs . rename ( taskOutput , finalOutputPath ) ) { if ( ! fs . delete ( finalOutputPath , true ) ) { throw new IOException ( "Failed to delete earlier output of task: " + attemptId ) ; } if ( ! fs . rename ( taskOutput , finalOutputPath ) ) { throw new IOException ( "Failed to save output of task: " + attemptId ) ; } } LOG . debug ( "Moved " + taskOutput + " to " + finalOutputPath ) ; } else if ( fs . getFileStatus ( taskOutput ) . isDir ( ) ) { FileStatus [ ] paths = fs . listStatus ( taskOutput ) ; Path finalOutputPath = getFinalPath ( jobOutputDir , taskOutput , workPath ) ; fs . mkdirs ( finalOutputPath ) ; if ( paths != null ) { for ( FileStatus path : paths ) { moveTaskOutputs ( context , fs , jobOutputDir , path . getPath ( ) ) ; } } } }
Move all of the files from the work directory to the final output