code: 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());
          }
        }
      }
    }
docs: Move all of the files from the work directory to the final output
  @param context the task context
  @param fs the output file system
  @param jobOutputDir the final output direcotry
  @param taskOutput the work path
  @throws IOException