@Override
    protected Path createInputFileListing(Job job) throws IOException {

      if (conf.get(NUMBER_OF_LEVELS_TO_PRESERVE_KEY) == null) {
        return super.createInputFileListing(job);
      }
      long totalBytesExpected = 0;
      int totalRecords = 0;
      Path fileListingPath = getFileListingPath();
      try (SequenceFile.Writer writer = getWriter(fileListingPath)) {
        List<Path> srcFiles = getSourceFiles();
        if (srcFiles.size() == 0) {
          return fileListingPath;
        }
        totalRecords = srcFiles.size();
        FileSystem fs = srcFiles.get(0).getFileSystem(conf);
        for (Path path : srcFiles) {
          FileStatus fst = fs.getFileStatus(path);
          totalBytesExpected += fst.getLen();
          Text key = getKey(path);
          writer.append(key, new CopyListingFileStatus(fst));
        }
        writer.close();

        // update jobs configuration

        Configuration cfg = job.getConfiguration();
        cfg.setLong(DistCpConstants.CONF_LABEL_TOTAL_BYTES_TO_BE_COPIED, totalBytesExpected);
        cfg.set(DistCpConstants.CONF_LABEL_LISTING_FILE_PATH, fileListingPath.toString());
        cfg.setLong(DistCpConstants.CONF_LABEL_TOTAL_NUMBER_OF_RECORDS, totalRecords);
      } catch (NoSuchFieldException | SecurityException | IllegalArgumentException
        | IllegalAccessException | NoSuchMethodException | ClassNotFoundException
        | InvocationTargetException e) {
        throw new IOException(e);
      }
      return fileListingPath;
    }
Extends DistCp for progress updating to backup system table during backup. Using DistCpV2
(MAPREDUCE-2765). Simply extend it and override execute() method to get the Job reference for
progress updating. Only the argument "src1, [src2, [...]] dst" is supported, no more DistCp
options.
hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/mapreduce/MapReduceBackupCopyJob.java