public Job createSubmittableJob(String[] args) throws IOException {
    if (!doCommandLine(args)) {
      return null;
    }

    String jobName = NAME + "_" + (tableName == null ? snapshot : tableName);
    Job job = Job.getInstance(getConf(), getConf().get(JOB_NAME_CONF_KEY, jobName));
    job.setJarByClass(CopyTable.class);
    Scan scan = new Scan();

    scan.setBatch(batch);
    scan.setCacheBlocks(false);

    if (cacheRow > 0) {
      scan.setCaching(cacheRow);
    } else {
      scan.setCaching(getConf().getInt(HConstants.HBASE_CLIENT_SCANNER_CACHING, 100));
    }

    scan.setTimeRange(startTime, endTime);

    if (allCells) {
      scan.setRaw(true);
    }
    if (shuffle) {
      job.getConfiguration().set(TableInputFormat.SHUFFLE_MAPS, "true");
    }
    if (versions >= 0) {
      scan.readVersions(versions);
    }

    if (startRow != null) {
      scan.withStartRow(Bytes.toBytesBinary(startRow));
    }

    if (stopRow != null) {
      scan.withStopRow(Bytes.toBytesBinary(stopRow));
    }

    if (families != null) {
      String[] fams = families.split(",");
      Map<String, String> cfRenameMap = new HashMap<>();
      for (String fam : fams) {
        String sourceCf;
        if (fam.contains(":")) {
          // fam looks like "sourceCfName:destCfName"
          String[] srcAndDest = fam.split(":", 2);
          sourceCf = srcAndDest[0];
          String destCf = srcAndDest[1];
          cfRenameMap.put(sourceCf, destCf);
        } else {
          // fam is just "sourceCf"
          sourceCf = fam;
        }
        scan.addFamily(Bytes.toBytes(sourceCf));
      }
      Import.configureCfRenaming(job.getConfiguration(), cfRenameMap);
    }
    job.setNumReduceTasks(0);

    if (bulkload) {
      initCopyTableMapperJob(job, scan);

      // We need to split the inputs by destination tables so that output of Map can be bulk-loaded.
      TableInputFormat.configureSplitTable(job, TableName.valueOf(dstTableName));

      bulkloadDir = generateUniqTempDir(false);
      LOG.info("HFiles will be stored at " + this.bulkloadDir);
      HFileOutputFormat2.setOutputPath(job, bulkloadDir);
      try (Connection conn = ConnectionFactory.createConnection(getConf());
        Admin admin = conn.getAdmin()) {
        HFileOutputFormat2.configureIncrementalLoadMap(job,
          admin.getDescriptor((TableName.valueOf(dstTableName))));
      }
    } else {
      initCopyTableMapperJob(job, scan);
      if (peerUri != null) {
        TableMapReduceUtil.initTableReducerJob(dstTableName, null, job, null, peerUri);
      } else if (peerAddress != null) {
        TableMapReduceUtil.initTableReducerJob(dstTableName, null, job, null, peerAddress);
      } else {
        TableMapReduceUtil.initTableReducerJob(dstTableName, null, job);
      }

    }

    return job;
  }
Sets up the actual job.
@param args The command line parameters.
@return The newly created job.
@throws IOException When setting up the job fails.
hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/CopyTable.java