public Job createSubmittableJob(String[] args) throws IOException {
    FileSystem fs = sourceHashDir.getFileSystem(getConf());
    if (!fs.exists(sourceHashDir)) {
      throw new IOException("Source hash dir not found: " + sourceHashDir);
    }

    Job job = Job.getInstance(getConf(),
      getConf().get("mapreduce.job.name", "syncTable_" + sourceTableName + "-" + targetTableName));
    Configuration jobConf = job.getConfiguration();
    if ("kerberos".equalsIgnoreCase(jobConf.get("hadoop.security.authentication"))) {
      TokenCache.obtainTokensForNamenodes(job.getCredentials(), new Path[] { sourceHashDir },
        getConf());
    }

    HashTable.TableHash tableHash = HashTable.TableHash.read(getConf(), sourceHashDir);
    LOG.info("Read source hash manifest: " + tableHash);
    LOG.info("Read " + tableHash.partitions.size() + " partition keys");
    if (!tableHash.tableName.equals(sourceTableName)) {
      LOG.warn("Table name mismatch - manifest indicates hash was taken from: "
        + tableHash.tableName + " but job is reading from: " + sourceTableName);
    }
    if (tableHash.numHashFiles != tableHash.partitions.size() + 1) {
      throw new RuntimeException("Hash data appears corrupt. The number of of hash files created"
        + " should be 1 more than the number of partition keys.  However, the manifest file "
        + " says numHashFiles=" + tableHash.numHashFiles + " but the number of partition keys"
        + " found in the partitions file is " + tableHash.partitions.size());
    }

    Path dataDir = new Path(sourceHashDir, HashTable.HASH_DATA_DIR);
    int dataSubdirCount = 0;
    for (FileStatus file : fs.listStatus(dataDir)) {
      if (file.getPath().getName().startsWith(HashTable.OUTPUT_DATA_FILE_PREFIX)) {
        dataSubdirCount++;
      }
    }

    if (dataSubdirCount != tableHash.numHashFiles) {
      throw new RuntimeException("Hash data appears corrupt. The number of of hash files created"
        + " should be 1 more than the number of partition keys.  However, the number of data dirs"
        + " found is " + dataSubdirCount + " but the number of partition keys"
        + " found in the partitions file is " + tableHash.partitions.size());
    }

    job.setJarByClass(HashTable.class);
    jobConf.set(SOURCE_HASH_DIR_CONF_KEY, sourceHashDir.toString());
    jobConf.set(SOURCE_TABLE_CONF_KEY, sourceTableName);
    jobConf.set(TARGET_TABLE_CONF_KEY, targetTableName);
    if (sourceUri != null) {
      jobConf.set(SOURCE_URI_CONF_KEY, sourceUri.toString());
      TableMapReduceUtil.initCredentialsForCluster(job, jobConf, sourceUri);
    } else if (sourceZkCluster != null) {
      jobConf.set(SOURCE_ZK_CLUSTER_CONF_KEY, sourceZkCluster);
      initCredentialsForHBase(sourceZkCluster, job);
    }
    if (targetUri != null) {
      jobConf.set(TARGET_URI_CONF_KEY, targetUri.toString());
      TableMapReduceUtil.initCredentialsForCluster(job, jobConf, targetUri);
    } else if (targetZkCluster != null) {
      jobConf.set(TARGET_ZK_CLUSTER_CONF_KEY, targetZkCluster);
      initCredentialsForHBase(targetZkCluster, job);
    }
    jobConf.setBoolean(DRY_RUN_CONF_KEY, dryRun);
    jobConf.setBoolean(DO_DELETES_CONF_KEY, doDeletes);
    jobConf.setBoolean(DO_PUTS_CONF_KEY, doPuts);
    jobConf.setBoolean(IGNORE_TIMESTAMPS, ignoreTimestamps);

    TableMapReduceUtil.initTableMapperJob(targetTableName, tableHash.initScan(), SyncMapper.class,
      null, null, job);

    job.setNumReduceTasks(0);

    if (dryRun) {
      job.setOutputFormatClass(NullOutputFormat.class);
    } else {
      // No reducers. Just write straight to table. Call initTableReducerJob
      // because it sets up the TableOutputFormat.
      if (targetUri != null) {
        TableMapReduceUtil.initTableReducerJob(targetTableName, null, job, null, targetUri);
      } else {
        TableMapReduceUtil.initTableReducerJob(targetTableName, null, job, null, targetZkCluster);
      }
      // would be nice to add an option for bulk load instead
    }

    return job;
  }
@deprecated Since 3.0.0, will be removed in 4.0.0 Use {@link #targetUri} instead.
hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/SyncTable.java