static void configureIncrementalLoad(Job job, List<TableInfo> multiTableInfo,
    Class<? extends OutputFormat<?, ?>> cls) throws IOException {
    Configuration conf = job.getConfiguration();
    job.setOutputKeyClass(ImmutableBytesWritable.class);
    job.setOutputValueClass(MapReduceExtendedCell.class);
    job.setOutputFormatClass(cls);

    if (multiTableInfo.stream().distinct().count() != multiTableInfo.size()) {
      throw new IllegalArgumentException("Duplicate entries found in TableInfo argument");
    }
    boolean writeMultipleTables = false;
    if (MultiTableHFileOutputFormat.class.equals(cls)) {
      writeMultipleTables = true;
      conf.setBoolean(MULTI_TABLE_HFILEOUTPUTFORMAT_CONF_KEY, true);
    }
    // Based on the configured map output class, set the correct reducer to properly
    // sort the incoming values.
    // TODO it would be nice to pick one or the other of these formats.
    if (
      KeyValue.class.equals(job.getMapOutputValueClass())
        || MapReduceExtendedCell.class.equals(job.getMapOutputValueClass())
    ) {
      job.setReducerClass(CellSortReducer.class);
    } else if (Put.class.equals(job.getMapOutputValueClass())) {
      job.setReducerClass(PutSortReducer.class);
    } else if (Text.class.equals(job.getMapOutputValueClass())) {
      job.setReducerClass(TextSortReducer.class);
    } else {
      LOG.warn("Unknown map output value type:" + job.getMapOutputValueClass());
    }

    mergeSerializations(conf);

    if (conf.getBoolean(LOCALITY_SENSITIVE_CONF_KEY, DEFAULT_LOCALITY_SENSITIVE)) {
      LOG.info("bulkload locality sensitive enabled");
    }

    /* Now get the region start keys for every table required */
    List<String> allTableNames = new ArrayList<>(multiTableInfo.size());
    List<RegionLocator> regionLocators = new ArrayList<>(multiTableInfo.size());
    List<TableDescriptor> tableDescriptors = new ArrayList<>(multiTableInfo.size());

    for (TableInfo tableInfo : multiTableInfo) {
      regionLocators.add(tableInfo.getRegionLocator());
      String tn = writeMultipleTables
        ? tableInfo.getRegionLocator().getName().getNameWithNamespaceInclAsString()
        : tableInfo.getRegionLocator().getName().getNameAsString();
      allTableNames.add(tn);
      tableDescriptors.add(tableInfo.getTableDescriptor());
    }
    // Record tablenames for creating writer by favored nodes, and decoding compression,
    // block size and other attributes of columnfamily per table
    conf.set(OUTPUT_TABLE_NAME_CONF_KEY,
      StringUtils.join(allTableNames, Bytes.toString(tableSeparator)));
    List<ImmutableBytesWritable> startKeys =
      getRegionStartKeys(regionLocators, writeMultipleTables);
    // Use table's region boundaries for TOP split points.
    LOG.info("Configuring " + startKeys.size() + " reduce partitions "
      + "to match current region count for all tables");
    job.setNumReduceTasks(startKeys.size());

    configurePartitioner(job, startKeys, writeMultipleTables);
    // Set compression algorithms based on column families

    conf.set(COMPRESSION_FAMILIES_CONF_KEY,
      serializeColumnFamilyAttribute(compressionDetails, tableDescriptors));
    conf.set(BLOCK_SIZE_FAMILIES_CONF_KEY,
      serializeColumnFamilyAttribute(blockSizeDetails, tableDescriptors));
    conf.set(BLOOM_TYPE_FAMILIES_CONF_KEY,
      serializeColumnFamilyAttribute(bloomTypeDetails, tableDescriptors));
    conf.set(BLOOM_PARAM_FAMILIES_CONF_KEY,
      serializeColumnFamilyAttribute(bloomParamDetails, tableDescriptors));
    conf.set(DATABLOCK_ENCODING_FAMILIES_CONF_KEY,
      serializeColumnFamilyAttribute(dataBlockEncodingDetails, tableDescriptors));

    TableMapReduceUtil.addDependencyJars(job);
    TableMapReduceUtil.initCredentials(job);
    LOG.info("Incremental output configured for tables: " + StringUtils.join(allTableNames, ","));
  }
Configure a MapReduce Job to perform an incremental load into the given table. This
<ul>
<li>Inspects the table to configure a total order partitioner</li>
<li>Uploads the partitions file to the cluster and adds it to the DistributedCache</li>
<li>Sets the number of reduce tasks to match the current number of regions</li>
<li>Sets the output key/value class to match HFileOutputFormat2's requirements</li>
<li>Sets the reducer up to perform the appropriate sorting (either KeyValueSortReducer or
PutSortReducer)</li>
</ul>
The user should be sure to set the map output value class to either KeyValue or Put before
running this function.
hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/HFileOutputFormat2.java