@Override
  public int run(String[] args) throws Exception {
    if (args.length < 2) {
      usage("Wrong number of arguments: " + args.length);
      return -1;
    }

    // When MAPPER_CONF_KEY is null, the user wants to use the provided TsvImporterMapper, so
    // perform validation on these additional args. When it's not null, user has provided their
    // own mapper, thus these validation are not relevant.
    // TODO: validation for TsvImporterMapper, not this tool. Move elsewhere.
    if (null == getConf().get(MAPPER_CONF_KEY)) {
      // Make sure columns are specified
      String[] columns = getConf().getStrings(COLUMNS_CONF_KEY);
      if (columns == null) {
        usage("No columns specified. Please specify with -D" + COLUMNS_CONF_KEY + "=...");
        return -1;
      }

      // Make sure they specify exactly one column as the row key
      int rowkeysFound = 0;
      for (String col : columns) {
        if (col.equals(TsvParser.ROWKEY_COLUMN_SPEC)) rowkeysFound++;
      }
      if (rowkeysFound != 1) {
        usage("Must specify exactly one column as " + TsvParser.ROWKEY_COLUMN_SPEC);
        return -1;
      }

      // Make sure we have at most one column as the timestamp key
      int tskeysFound = 0;
      for (String col : columns) {
        if (col.equals(TsvParser.TIMESTAMPKEY_COLUMN_SPEC)) tskeysFound++;
      }
      if (tskeysFound > 1) {
        usage("Must specify at most one column as " + TsvParser.TIMESTAMPKEY_COLUMN_SPEC);
        return -1;
      }

      int attrKeysFound = 0;
      for (String col : columns) {
        if (col.equals(TsvParser.ATTRIBUTES_COLUMN_SPEC)) attrKeysFound++;
      }
      if (attrKeysFound > 1) {
        usage("Must specify at most one column as " + TsvParser.ATTRIBUTES_COLUMN_SPEC);
        return -1;
      }

      // Make sure one or more columns are specified excluding rowkey and
      // timestamp key
      if (columns.length - (rowkeysFound + tskeysFound + attrKeysFound) < 1) {
        usage(
          "One or more columns in addition to the row key and timestamp(optional) are required");
        return -1;
      }
    }

    // If timestamp option is not specified, use current system time.
    long timstamp = getConf().getLong(TIMESTAMP_CONF_KEY, EnvironmentEdgeManager.currentTime());

    // Set it back to replace invalid timestamp (non-numeric) with current
    // system time
    getConf().setLong(TIMESTAMP_CONF_KEY, timstamp);

    synchronized (ImportTsv.class) {
      DRY_RUN_TABLE_CREATED = false;
    }
    Job job = createSubmittableJob(getConf(), args);
    boolean success = job.waitForCompletion(true);
    boolean delete = false;
    synchronized (ImportTsv.class) {
      delete = DRY_RUN_TABLE_CREATED;
    }
    if (delete) {
      deleteTable(getConf(), args);
    }
    return success ? 0 : 1;
  }
Sets up the actual job.
@param conf The current configuration.
@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/ImportTsv.java