public static Job createSubmittableJob(Configuration conf, String[] args) throws IOException {
    TableName tableName = TableName.valueOf(args[0]);
    conf.set(TABLE_NAME, tableName.getNameAsString());
    Path inputDir = new Path(args[1]);
    Job job = Job.getInstance(conf, conf.get(JOB_NAME_CONF_KEY, NAME + "_" + tableName));
    job.setJarByClass(Importer.class);
    FileInputFormat.setInputPaths(job, inputDir);
    job.setInputFormatClass(SequenceFileInputFormat.class);
    String hfileOutPath = conf.get(BULK_OUTPUT_CONF_KEY);

    // make sure we get the filter in the jars
    try {
      Class<? extends Filter> filter = conf.getClass(FILTER_CLASS_CONF_KEY, null, Filter.class);
      if (filter != null) {
        TableMapReduceUtil.addDependencyJarsForClasses(conf, filter);
      }
    } catch (Exception e) {
      throw new IOException(e);
    }

    if (hfileOutPath != null && conf.getBoolean(HAS_LARGE_RESULT, false)) {
      LOG.info("Use Large Result!!");
      try (Connection conn = ConnectionFactory.createConnection(conf);
        Table table = conn.getTable(tableName);
        RegionLocator regionLocator = conn.getRegionLocator(tableName)) {
        HFileOutputFormat2.configureIncrementalLoad(job, table.getDescriptor(), regionLocator);
        job.setMapperClass(CellSortImporter.class);
        job.setReducerClass(CellReducer.class);
        Path outputDir = new Path(hfileOutPath);
        FileOutputFormat.setOutputPath(job, outputDir);
        job.setMapOutputKeyClass(CellWritableComparable.class);
        job.setMapOutputValueClass(MapReduceExtendedCell.class);
        job.getConfiguration().setClass("mapreduce.job.output.key.comparator.class",
          CellWritableComparable.CellWritableComparator.class, RawComparator.class);
        Path partitionsPath =
          new Path(TotalOrderPartitioner.getPartitionFile(job.getConfiguration()));
        FileSystem fs = FileSystem.get(job.getConfiguration());
        fs.deleteOnExit(partitionsPath);
        job.setPartitionerClass(CellWritableComparablePartitioner.class);
        job.setNumReduceTasks(regionLocator.getStartKeys().length);
        TableMapReduceUtil.addDependencyJarsForClasses(job.getConfiguration(),
          org.apache.hbase.thirdparty.com.google.common.base.Preconditions.class);
      }
    } else if (hfileOutPath != null) {
      LOG.info("writing to hfiles for bulk load.");
      job.setMapperClass(CellImporter.class);
      try (Connection conn = ConnectionFactory.createConnection(conf);
        Table table = conn.getTable(tableName);
        RegionLocator regionLocator = conn.getRegionLocator(tableName)) {
        job.setReducerClass(CellSortReducer.class);
        Path outputDir = new Path(hfileOutPath);
        FileOutputFormat.setOutputPath(job, outputDir);
        job.setMapOutputKeyClass(ImmutableBytesWritable.class);
        job.setMapOutputValueClass(MapReduceExtendedCell.class);
        HFileOutputFormat2.configureIncrementalLoad(job, table.getDescriptor(), regionLocator);
        TableMapReduceUtil.addDependencyJarsForClasses(job.getConfiguration(),
          org.apache.hbase.thirdparty.com.google.common.base.Preconditions.class);
      }
    } else {
      LOG.info("writing directly to table from Mapper.");
      // No reducers. Just write straight to table. Call initTableReducerJob
      // because it sets up the TableOutputFormat.
      job.setMapperClass(Importer.class);
      TableMapReduceUtil.initTableReducerJob(tableName.getNameAsString(), null, job);
      job.setNumReduceTasks(0);
    }
    return job;
  }
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/Import.java