public int run(String[] args) throws IOException, InterruptedException {
    // TODO make family and table optional
    if (args.length != 3) {
      printUsage();
      return 1;
    }
    final String output = args[0];
    final String tableName = args[1];
    final String familyName = args[2];
    final long reportStartTime = EnvironmentEdgeManager.currentTime();
    Configuration conf = getConf();
    try {
      FileSystem fs = FileSystem.get(conf);
      // check whether the current user is the same one with the owner of hbase root
      String currentUserName = UserGroupInformation.getCurrentUser().getShortUserName();
      FileStatus[] hbaseRootFileStat = fs.listStatus(new Path(conf.get(HConstants.HBASE_DIR)));
      if (hbaseRootFileStat.length > 0) {
        String owner = hbaseRootFileStat[0].getOwner();
        if (!owner.equals(currentUserName)) {
          String errorMsg =
            "The current user[" + currentUserName + "] does not have hbase root credentials."
              + " If this job fails due to an inability to read HBase's internal directories, "
              + "you will need to rerun as a user with sufficient permissions. The HBase superuser "
              + "is a safe choice.";
          LOG.warn(errorMsg);
        }
      } else {
        LOG.error("The passed configs point to an HBase dir does not exist: {}",
          conf.get(HConstants.HBASE_DIR));
        throw new IOException("The target HBase does not exist");
      }

      byte[] family;
      int maxVersions;
      TableName tn = TableName.valueOf(tableName);
      try (Connection connection = ConnectionFactory.createConnection(conf);
        Admin admin = connection.getAdmin()) {
        TableDescriptor htd = admin.getDescriptor(tn);
        ColumnFamilyDescriptor hcd = htd.getColumnFamily(Bytes.toBytes(familyName));
        if (hcd == null || !hcd.isMobEnabled()) {
          throw new IOException("Column family " + familyName + " is not a MOB column family");
        }
        family = hcd.getName();
        maxVersions = hcd.getMaxVersions();
      }

      String id = getClass().getSimpleName() + UUID.randomUUID().toString().replace("-", "");
      Job job = null;
      Scan scan = new Scan();
      scan.addFamily(family);
      // Do not retrieve the mob data when scanning
      scan.setAttribute(MobConstants.MOB_SCAN_RAW, Bytes.toBytes(Boolean.TRUE));
      scan.setAttribute(MobConstants.MOB_SCAN_REF_ONLY, Bytes.toBytes(Boolean.TRUE));
      // If a scanner caching value isn't set, pick a smaller default since we know we're doing
      // a full table scan and don't want to impact other clients badly.
      scan.setCaching(conf.getInt(HConstants.HBASE_CLIENT_SCANNER_CACHING, 10000));
      scan.setCacheBlocks(false);
      scan.readVersions(maxVersions);
      conf.set(REPORT_JOB_ID, id);

      job = Job.getInstance(conf);
      job.setJarByClass(getClass());
      TableMapReduceUtil.initTableMapperJob(tn, scan, MobRefMapper.class, Text.class,
        ImmutableBytesWritable.class, job);

      job.setReducerClass(MobRefReducer.class);
      job.setOutputFormatClass(TextOutputFormat.class);
      TextOutputFormat.setOutputPath(job, new Path(output));

      job.setJobName(getClass().getSimpleName() + "-" + tn + "-" + familyName);
      // for use in the reducer. easier than re-parsing it out of the scan string.
      job.getConfiguration().set(TableInputFormat.SCAN_COLUMN_FAMILY, familyName);

      // Use when we start this job as the base point for file "recency".
      job.getConfiguration().setLong(REPORT_START_DATETIME, reportStartTime);

      if (job.waitForCompletion(true)) {
        LOG.info("Finished creating report for '{}', family='{}'", tn, familyName);
      } else {
        System.err.println("Job was not successful");
        return 3;
      }
      return 0;

    } catch (ClassNotFoundException | RuntimeException | IOException | InterruptedException e) {
      System.err.println("Job aborted due to exception " + e);
      return 2; // job failed
    }
  }
Main method for the tool.
@return 0 if success, 1 for bad args. 2 if job aborted with an exception, 3 if mr job was
        unsuccessful
hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mob/mapreduce/MobRefReporter.java