001/*
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018package org.apache.hadoop.hbase.mapreduce;
019
020import static org.apache.hadoop.hbase.regionserver.HStoreFile.BULKLOAD_TASK_KEY;
021import static org.apache.hadoop.hbase.regionserver.HStoreFile.BULKLOAD_TIME_KEY;
022import static org.apache.hadoop.hbase.regionserver.HStoreFile.EXCLUDE_FROM_MINOR_COMPACTION_KEY;
023import static org.apache.hadoop.hbase.regionserver.HStoreFile.MAJOR_COMPACTION_KEY;
024
025import java.io.IOException;
026import java.io.UnsupportedEncodingException;
027import java.net.InetSocketAddress;
028import java.net.URLDecoder;
029import java.net.URLEncoder;
030import java.nio.charset.Charset;
031import java.util.ArrayList;
032import java.util.Arrays;
033import java.util.List;
034import java.util.Map;
035import java.util.Set;
036import java.util.TreeMap;
037import java.util.TreeSet;
038import java.util.UUID;
039import java.util.function.Function;
040import java.util.stream.Collectors;
041import org.apache.commons.lang3.StringUtils;
042import org.apache.hadoop.conf.Configuration;
043import org.apache.hadoop.fs.FileSystem;
044import org.apache.hadoop.fs.Path;
045import org.apache.hadoop.hbase.Cell;
046import org.apache.hadoop.hbase.CellUtil;
047import org.apache.hadoop.hbase.HConstants;
048import org.apache.hadoop.hbase.HRegionLocation;
049import org.apache.hadoop.hbase.KeyValue;
050import org.apache.hadoop.hbase.PrivateCellUtil;
051import org.apache.hadoop.hbase.TableName;
052import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
053import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
054import org.apache.hadoop.hbase.client.Connection;
055import org.apache.hadoop.hbase.client.ConnectionFactory;
056import org.apache.hadoop.hbase.client.Put;
057import org.apache.hadoop.hbase.client.RegionLocator;
058import org.apache.hadoop.hbase.client.Table;
059import org.apache.hadoop.hbase.client.TableDescriptor;
060import org.apache.hadoop.hbase.fs.HFileSystem;
061import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
062import org.apache.hadoop.hbase.io.compress.Compression;
063import org.apache.hadoop.hbase.io.compress.Compression.Algorithm;
064import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
065import org.apache.hadoop.hbase.io.hfile.CacheConfig;
066import org.apache.hadoop.hbase.io.hfile.HFile;
067import org.apache.hadoop.hbase.io.hfile.HFileContext;
068import org.apache.hadoop.hbase.io.hfile.HFileContextBuilder;
069import org.apache.hadoop.hbase.io.hfile.HFileWriterImpl;
070import org.apache.hadoop.hbase.regionserver.BloomType;
071import org.apache.hadoop.hbase.regionserver.HStore;
072import org.apache.hadoop.hbase.regionserver.StoreFileWriter;
073import org.apache.hadoop.hbase.util.BloomFilterUtil;
074import org.apache.hadoop.hbase.util.Bytes;
075import org.apache.hadoop.hbase.util.CommonFSUtils;
076import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
077import org.apache.hadoop.hbase.util.MapReduceExtendedCell;
078import org.apache.hadoop.io.NullWritable;
079import org.apache.hadoop.io.SequenceFile;
080import org.apache.hadoop.io.Text;
081import org.apache.hadoop.mapreduce.Job;
082import org.apache.hadoop.mapreduce.OutputCommitter;
083import org.apache.hadoop.mapreduce.OutputFormat;
084import org.apache.hadoop.mapreduce.RecordWriter;
085import org.apache.hadoop.mapreduce.TaskAttemptContext;
086import org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter;
087import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
088import org.apache.hadoop.mapreduce.lib.partition.TotalOrderPartitioner;
089import org.apache.yetus.audience.InterfaceAudience;
090import org.slf4j.Logger;
091import org.slf4j.LoggerFactory;
092
093import org.apache.hbase.thirdparty.com.google.common.annotations.VisibleForTesting;
094
095/**
096 * Writes HFiles. Passed Cells must arrive in order.
097 * Writes current time as the sequence id for the file. Sets the major compacted
098 * attribute on created @{link {@link HFile}s. Calling write(null,null) will forcibly roll
099 * all HFiles being written.
100 * <p>
101 * Using this class as part of a MapReduce job is best done
102 * using {@link #configureIncrementalLoad(Job, TableDescriptor, RegionLocator)}.
103 */
104@InterfaceAudience.Public
105public class HFileOutputFormat2
106    extends FileOutputFormat<ImmutableBytesWritable, Cell> {
107  private static final Logger LOG = LoggerFactory.getLogger(HFileOutputFormat2.class);
108  static class TableInfo {
109    private TableDescriptor tableDesctiptor;
110    private RegionLocator regionLocator;
111
112    public TableInfo(TableDescriptor tableDesctiptor, RegionLocator regionLocator) {
113      this.tableDesctiptor = tableDesctiptor;
114      this.regionLocator = regionLocator;
115    }
116
117    public TableDescriptor getTableDescriptor() {
118      return tableDesctiptor;
119    }
120
121    public RegionLocator getRegionLocator() {
122      return regionLocator;
123    }
124  }
125
126  protected static final byte[] tableSeparator = Bytes.toBytes(";");
127
128  protected static byte[] combineTableNameSuffix(byte[] tableName, byte[] suffix) {
129    return Bytes.add(tableName, tableSeparator, suffix);
130  }
131
132  // The following constants are private since these are used by
133  // HFileOutputFormat2 to internally transfer data between job setup and
134  // reducer run using conf.
135  // These should not be changed by the client.
136  static final String COMPRESSION_FAMILIES_CONF_KEY =
137      "hbase.hfileoutputformat.families.compression";
138  static final String BLOOM_TYPE_FAMILIES_CONF_KEY =
139      "hbase.hfileoutputformat.families.bloomtype";
140  static final String BLOOM_PARAM_FAMILIES_CONF_KEY =
141      "hbase.hfileoutputformat.families.bloomparam";
142  static final String BLOCK_SIZE_FAMILIES_CONF_KEY =
143      "hbase.mapreduce.hfileoutputformat.blocksize";
144  static final String DATABLOCK_ENCODING_FAMILIES_CONF_KEY =
145      "hbase.mapreduce.hfileoutputformat.families.datablock.encoding";
146
147  // This constant is public since the client can modify this when setting
148  // up their conf object and thus refer to this symbol.
149  // It is present for backwards compatibility reasons. Use it only to
150  // override the auto-detection of datablock encoding and compression.
151  public static final String DATABLOCK_ENCODING_OVERRIDE_CONF_KEY =
152      "hbase.mapreduce.hfileoutputformat.datablock.encoding";
153  public static final String COMPRESSION_OVERRIDE_CONF_KEY =
154      "hbase.mapreduce.hfileoutputformat.compression";
155
156  /**
157   * Keep locality while generating HFiles for bulkload. See HBASE-12596
158   */
159  public static final String LOCALITY_SENSITIVE_CONF_KEY =
160      "hbase.bulkload.locality.sensitive.enabled";
161  private static final boolean DEFAULT_LOCALITY_SENSITIVE = true;
162  static final String OUTPUT_TABLE_NAME_CONF_KEY =
163      "hbase.mapreduce.hfileoutputformat.table.name";
164  static final String MULTI_TABLE_HFILEOUTPUTFORMAT_CONF_KEY =
165          "hbase.mapreduce.use.multi.table.hfileoutputformat";
166
167  public static final String STORAGE_POLICY_PROPERTY = HStore.BLOCK_STORAGE_POLICY_KEY;
168  public static final String STORAGE_POLICY_PROPERTY_CF_PREFIX = STORAGE_POLICY_PROPERTY + ".";
169
170  @Override
171  public RecordWriter<ImmutableBytesWritable, Cell> getRecordWriter(
172      final TaskAttemptContext context) throws IOException, InterruptedException {
173    return createRecordWriter(context, this.getOutputCommitter(context));
174  }
175
176  protected static byte[] getTableNameSuffixedWithFamily(byte[] tableName, byte[] family) {
177    return combineTableNameSuffix(tableName, family);
178  }
179
180  static <V extends Cell> RecordWriter<ImmutableBytesWritable, V> createRecordWriter(
181    final TaskAttemptContext context, final OutputCommitter committer) throws IOException {
182
183    // Get the path of the temporary output file
184    final Path outputDir = ((FileOutputCommitter)committer).getWorkPath();
185    final Configuration conf = context.getConfiguration();
186    final boolean writeMultipleTables =
187      conf.getBoolean(MULTI_TABLE_HFILEOUTPUTFORMAT_CONF_KEY, false);
188    final String writeTableNames = conf.get(OUTPUT_TABLE_NAME_CONF_KEY);
189    if (writeTableNames == null || writeTableNames.isEmpty()) {
190      throw new IllegalArgumentException("" + OUTPUT_TABLE_NAME_CONF_KEY + " cannot be empty");
191    }
192    final FileSystem fs = outputDir.getFileSystem(conf);
193    // These configs. are from hbase-*.xml
194    final long maxsize = conf.getLong(HConstants.HREGION_MAX_FILESIZE,
195        HConstants.DEFAULT_MAX_FILE_SIZE);
196    // Invented config.  Add to hbase-*.xml if other than default compression.
197    final String defaultCompressionStr = conf.get("hfile.compression",
198        Compression.Algorithm.NONE.getName());
199    final Algorithm defaultCompression = HFileWriterImpl.compressionByName(defaultCompressionStr);
200    String compressionStr = conf.get(COMPRESSION_OVERRIDE_CONF_KEY);
201    final Algorithm overriddenCompression = compressionStr != null ?
202      Compression.getCompressionAlgorithmByName(compressionStr): null;
203    final boolean compactionExclude = conf.getBoolean(
204        "hbase.mapreduce.hfileoutputformat.compaction.exclude", false);
205    final Set<String> allTableNames = Arrays.stream(writeTableNames.split(
206            Bytes.toString(tableSeparator))).collect(Collectors.toSet());
207
208    // create a map from column family to the compression algorithm
209    final Map<byte[], Algorithm> compressionMap = createFamilyCompressionMap(conf);
210    final Map<byte[], BloomType> bloomTypeMap = createFamilyBloomTypeMap(conf);
211    final Map<byte[], String> bloomParamMap = createFamilyBloomParamMap(conf);
212    final Map<byte[], Integer> blockSizeMap = createFamilyBlockSizeMap(conf);
213
214    String dataBlockEncodingStr = conf.get(DATABLOCK_ENCODING_OVERRIDE_CONF_KEY);
215    final Map<byte[], DataBlockEncoding> datablockEncodingMap
216        = createFamilyDataBlockEncodingMap(conf);
217    final DataBlockEncoding overriddenEncoding = dataBlockEncodingStr != null ?
218      DataBlockEncoding.valueOf(dataBlockEncodingStr) : null;
219
220    return new RecordWriter<ImmutableBytesWritable, V>() {
221      // Map of families to writers and how much has been output on the writer.
222      private final Map<byte[], WriterLength> writers = new TreeMap<>(Bytes.BYTES_COMPARATOR);
223      private final Map<byte[], byte[]> previousRows = new TreeMap<>(Bytes.BYTES_COMPARATOR);
224      private final long now = EnvironmentEdgeManager.currentTime();
225      private byte[] tableNameBytes = writeMultipleTables ? null : Bytes.toBytes(writeTableNames);
226
227      @Override
228      public void write(ImmutableBytesWritable row, V cell) throws IOException {
229        Cell kv = cell;
230        // null input == user explicitly wants to flush
231        if (row == null && kv == null) {
232          rollWriters(null);
233          return;
234        }
235
236        byte[] rowKey = CellUtil.cloneRow(kv);
237        int length = (PrivateCellUtil.estimatedSerializedSizeOf(kv)) - Bytes.SIZEOF_INT;
238        byte[] family = CellUtil.cloneFamily(kv);
239        if (writeMultipleTables) {
240          tableNameBytes = MultiTableHFileOutputFormat.getTableName(row.get());
241          tableNameBytes = TableName.valueOf(tableNameBytes).getNameWithNamespaceInclAsString()
242              .getBytes(Charset.defaultCharset());
243          if (!allTableNames.contains(Bytes.toString(tableNameBytes))) {
244            throw new IllegalArgumentException("TableName " + Bytes.toString(tableNameBytes) +
245              " not expected");
246          }
247        }
248        byte[] tableAndFamily = getTableNameSuffixedWithFamily(tableNameBytes, family);
249
250        WriterLength wl = this.writers.get(tableAndFamily);
251
252        // If this is a new column family, verify that the directory exists
253        if (wl == null) {
254          Path writerPath = null;
255          if (writeMultipleTables) {
256            Path tableRelPath = getTableRelativePath(tableNameBytes);
257            writerPath = new Path(outputDir, new Path(tableRelPath, Bytes.toString(family)));
258          } else {
259            writerPath = new Path(outputDir, Bytes.toString(family));
260          }
261          fs.mkdirs(writerPath);
262          configureStoragePolicy(conf, fs, tableAndFamily, writerPath);
263        }
264
265        // This can only happen once a row is finished though
266        if (wl != null && wl.written + length >= maxsize
267                && Bytes.compareTo(this.previousRows.get(family), rowKey) != 0) {
268          rollWriters(wl);
269        }
270
271        // create a new WAL writer, if necessary
272        if (wl == null || wl.writer == null) {
273          InetSocketAddress[] favoredNodes = null;
274          if (conf.getBoolean(LOCALITY_SENSITIVE_CONF_KEY, DEFAULT_LOCALITY_SENSITIVE)) {
275            HRegionLocation loc = null;
276            String tableName = Bytes.toString(tableNameBytes);
277            if (tableName != null) {
278              try (Connection connection = ConnectionFactory.createConnection(conf);
279                  RegionLocator locator =
280                      connection.getRegionLocator(TableName.valueOf(tableName))) {
281                loc = locator.getRegionLocation(rowKey);
282              } catch (Throwable e) {
283                LOG.warn("Something wrong locating rowkey {} in {}", Bytes.toString(rowKey),
284                  tableName, e);
285                loc = null;
286              }
287            }
288            if (null == loc) {
289              LOG.trace("Failed get of location, use default writer {}", Bytes.toString(rowKey));
290            } else {
291              LOG.debug("First rowkey: [{}]", Bytes.toString(rowKey));
292              InetSocketAddress initialIsa =
293                  new InetSocketAddress(loc.getHostname(), loc.getPort());
294              if (initialIsa.isUnresolved()) {
295                LOG.trace("Failed resolve address {}, use default writer", loc.getHostnamePort());
296              } else {
297                LOG.debug("Use favored nodes writer: {}", initialIsa.getHostString());
298                favoredNodes = new InetSocketAddress[] { initialIsa };
299              }
300            }
301          }
302          wl = getNewWriter(tableNameBytes, family, conf, favoredNodes);
303
304        }
305
306        // we now have the proper WAL writer. full steam ahead
307        PrivateCellUtil.updateLatestStamp(cell, this.now);
308        wl.writer.append(kv);
309        wl.written += length;
310
311        // Copy the row so we know when a row transition.
312        this.previousRows.put(family, rowKey);
313      }
314
315      private Path getTableRelativePath(byte[] tableNameBytes) {
316        String tableName = Bytes.toString(tableNameBytes);
317        String[] tableNameParts = tableName.split(":");
318        Path tableRelPath = new Path(tableNameParts[0]);
319        if (tableNameParts.length > 1) {
320          tableRelPath = new Path(tableRelPath, tableNameParts[1]);
321        }
322        return tableRelPath;
323      }
324      private void rollWriters(WriterLength writerLength) throws IOException {
325        if (writerLength != null) {
326          closeWriter(writerLength);
327        } else {
328          for (WriterLength wl : this.writers.values()) {
329            closeWriter(wl);
330          }
331        }
332      }
333
334      private void closeWriter(WriterLength wl) throws IOException {
335        if (wl.writer != null) {
336          LOG.info("Writer=" + wl.writer.getPath() +
337            ((wl.written == 0)? "": ", wrote=" + wl.written));
338          close(wl.writer);
339          wl.writer = null;
340        }
341        wl.written = 0;
342      }
343
344      /*
345       * Create a new StoreFile.Writer.
346       * @return A WriterLength, containing a new StoreFile.Writer.
347       */
348      @edu.umd.cs.findbugs.annotations.SuppressWarnings(value="BX_UNBOXING_IMMEDIATELY_REBOXED",
349          justification="Not important")
350      private WriterLength getNewWriter(byte[] tableName, byte[] family, Configuration conf,
351          InetSocketAddress[] favoredNodes) throws IOException {
352        byte[] tableAndFamily = getTableNameSuffixedWithFamily(tableName, family);
353        Path familydir = new Path(outputDir, Bytes.toString(family));
354        if (writeMultipleTables) {
355          familydir = new Path(outputDir,
356            new Path(getTableRelativePath(tableName), Bytes.toString(family)));
357        }
358        WriterLength wl = new WriterLength();
359        Algorithm compression = overriddenCompression;
360        compression = compression == null ? compressionMap.get(tableAndFamily) : compression;
361        compression = compression == null ? defaultCompression : compression;
362        BloomType bloomType = bloomTypeMap.get(tableAndFamily);
363        bloomType = bloomType == null ? BloomType.NONE : bloomType;
364        String bloomParam = bloomParamMap.get(tableAndFamily);
365        if (bloomType == BloomType.ROWPREFIX_FIXED_LENGTH) {
366          conf.set(BloomFilterUtil.PREFIX_LENGTH_KEY, bloomParam);
367        }
368        Integer blockSize = blockSizeMap.get(tableAndFamily);
369        blockSize = blockSize == null ? HConstants.DEFAULT_BLOCKSIZE : blockSize;
370        DataBlockEncoding encoding = overriddenEncoding;
371        encoding = encoding == null ? datablockEncodingMap.get(tableAndFamily) : encoding;
372        encoding = encoding == null ? DataBlockEncoding.NONE : encoding;
373        HFileContextBuilder contextBuilder = new HFileContextBuilder().withCompression(compression)
374            .withDataBlockEncoding(encoding).withChecksumType(HStore.getChecksumType(conf))
375            .withBytesPerCheckSum(HStore.getBytesPerChecksum(conf)).withBlockSize(blockSize)
376            .withColumnFamily(family).withTableName(tableName);
377
378        if (HFile.getFormatVersion(conf) >= HFile.MIN_FORMAT_VERSION_WITH_TAGS) {
379          contextBuilder.withIncludesTags(true);
380        }
381
382        HFileContext hFileContext = contextBuilder.build();
383        if (null == favoredNodes) {
384          wl.writer = new StoreFileWriter.Builder(conf, CacheConfig.DISABLED, fs)
385            .withOutputDir(familydir).withBloomType(bloomType)
386            .withFileContext(hFileContext).build();
387        } else {
388          wl.writer = new StoreFileWriter.Builder(conf, CacheConfig.DISABLED, new HFileSystem(fs))
389            .withOutputDir(familydir).withBloomType(bloomType)
390            .withFileContext(hFileContext).withFavoredNodes(favoredNodes).build();
391        }
392
393        this.writers.put(tableAndFamily, wl);
394        return wl;
395      }
396
397      private void close(final StoreFileWriter w) throws IOException {
398        if (w != null) {
399          w.appendFileInfo(BULKLOAD_TIME_KEY, Bytes.toBytes(System.currentTimeMillis()));
400          w.appendFileInfo(BULKLOAD_TASK_KEY, Bytes.toBytes(context.getTaskAttemptID().toString()));
401          w.appendFileInfo(MAJOR_COMPACTION_KEY, Bytes.toBytes(true));
402          w.appendFileInfo(EXCLUDE_FROM_MINOR_COMPACTION_KEY, Bytes.toBytes(compactionExclude));
403          w.appendTrackedTimestampsToMetadata();
404          w.close();
405        }
406      }
407
408      @Override
409      public void close(TaskAttemptContext c) throws IOException, InterruptedException {
410        for (WriterLength wl: this.writers.values()) {
411          close(wl.writer);
412        }
413      }
414    };
415  }
416
417  /**
418   * Configure block storage policy for CF after the directory is created.
419   */
420  static void configureStoragePolicy(final Configuration conf, final FileSystem fs,
421      byte[] tableAndFamily, Path cfPath) {
422    if (null == conf || null == fs || null == tableAndFamily || null == cfPath) {
423      return;
424    }
425
426    String policy =
427        conf.get(STORAGE_POLICY_PROPERTY_CF_PREFIX + Bytes.toString(tableAndFamily),
428          conf.get(STORAGE_POLICY_PROPERTY));
429    CommonFSUtils.setStoragePolicy(fs, cfPath, policy);
430  }
431
432  /*
433   * Data structure to hold a Writer and amount of data written on it.
434   */
435  static class WriterLength {
436    long written = 0;
437    StoreFileWriter writer = null;
438  }
439
440  /**
441   * Return the start keys of all of the regions in this table,
442   * as a list of ImmutableBytesWritable.
443   */
444  private static List<ImmutableBytesWritable> getRegionStartKeys(List<RegionLocator> regionLocators,
445                                                                 boolean writeMultipleTables)
446          throws IOException {
447
448    ArrayList<ImmutableBytesWritable> ret = new ArrayList<>();
449    for(RegionLocator regionLocator : regionLocators) {
450      TableName tableName = regionLocator.getName();
451      LOG.info("Looking up current regions for table " + tableName);
452      byte[][] byteKeys = regionLocator.getStartKeys();
453      for (byte[] byteKey : byteKeys) {
454        byte[] fullKey = byteKey; //HFileOutputFormat2 use case
455        if (writeMultipleTables) {
456          //MultiTableHFileOutputFormat use case
457          fullKey = combineTableNameSuffix(tableName.getName(), byteKey);
458        }
459        if (LOG.isDebugEnabled()) {
460          LOG.debug("SplitPoint startkey for " + tableName + ": " + Bytes.toStringBinary(fullKey));
461        }
462        ret.add(new ImmutableBytesWritable(fullKey));
463      }
464    }
465    return ret;
466  }
467
468  /**
469   * Write out a {@link SequenceFile} that can be read by
470   * {@link TotalOrderPartitioner} that contains the split points in startKeys.
471   */
472  @SuppressWarnings("deprecation")
473  private static void writePartitions(Configuration conf, Path partitionsPath,
474      List<ImmutableBytesWritable> startKeys, boolean writeMultipleTables) throws IOException {
475    LOG.info("Writing partition information to " + partitionsPath);
476    if (startKeys.isEmpty()) {
477      throw new IllegalArgumentException("No regions passed");
478    }
479
480    // We're generating a list of split points, and we don't ever
481    // have keys < the first region (which has an empty start key)
482    // so we need to remove it. Otherwise we would end up with an
483    // empty reducer with index 0
484    TreeSet<ImmutableBytesWritable> sorted = new TreeSet<>(startKeys);
485    ImmutableBytesWritable first = sorted.first();
486    if (writeMultipleTables) {
487      first =
488        new ImmutableBytesWritable(MultiTableHFileOutputFormat.getSuffix(sorted.first().get()));
489    }
490    if (!first.equals(HConstants.EMPTY_BYTE_ARRAY)) {
491      throw new IllegalArgumentException(
492          "First region of table should have empty start key. Instead has: "
493          + Bytes.toStringBinary(first.get()));
494    }
495    sorted.remove(sorted.first());
496
497    // Write the actual file
498    FileSystem fs = partitionsPath.getFileSystem(conf);
499    SequenceFile.Writer writer = SequenceFile.createWriter(
500      fs, conf, partitionsPath, ImmutableBytesWritable.class,
501      NullWritable.class);
502
503    try {
504      for (ImmutableBytesWritable startKey : sorted) {
505        writer.append(startKey, NullWritable.get());
506      }
507    } finally {
508      writer.close();
509    }
510  }
511
512  /**
513   * Configure a MapReduce Job to perform an incremental load into the given
514   * table. This
515   * <ul>
516   *   <li>Inspects the table to configure a total order partitioner</li>
517   *   <li>Uploads the partitions file to the cluster and adds it to the DistributedCache</li>
518   *   <li>Sets the number of reduce tasks to match the current number of regions</li>
519   *   <li>Sets the output key/value class to match HFileOutputFormat2's requirements</li>
520   *   <li>Sets the reducer up to perform the appropriate sorting (either KeyValueSortReducer or
521   *     PutSortReducer)</li>
522   * </ul>
523   * The user should be sure to set the map output value class to either KeyValue or Put before
524   * running this function.
525   */
526  public static void configureIncrementalLoad(Job job, Table table, RegionLocator regionLocator)
527      throws IOException {
528    configureIncrementalLoad(job, table.getDescriptor(), regionLocator);
529  }
530
531  /**
532   * Configure a MapReduce Job to perform an incremental load into the given
533   * table. This
534   * <ul>
535   *   <li>Inspects the table to configure a total order partitioner</li>
536   *   <li>Uploads the partitions file to the cluster and adds it to the DistributedCache</li>
537   *   <li>Sets the number of reduce tasks to match the current number of regions</li>
538   *   <li>Sets the output key/value class to match HFileOutputFormat2's requirements</li>
539   *   <li>Sets the reducer up to perform the appropriate sorting (either KeyValueSortReducer or
540   *     PutSortReducer)</li>
541   * </ul>
542   * The user should be sure to set the map output value class to either KeyValue or Put before
543   * running this function.
544   */
545  public static void configureIncrementalLoad(Job job, TableDescriptor tableDescriptor,
546      RegionLocator regionLocator) throws IOException {
547    ArrayList<TableInfo> singleTableInfo = new ArrayList<>();
548    singleTableInfo.add(new TableInfo(tableDescriptor, regionLocator));
549    configureIncrementalLoad(job, singleTableInfo, HFileOutputFormat2.class);
550  }
551
552  static void configureIncrementalLoad(Job job, List<TableInfo> multiTableInfo,
553      Class<? extends OutputFormat<?, ?>> cls) throws IOException {
554    Configuration conf = job.getConfiguration();
555    job.setOutputKeyClass(ImmutableBytesWritable.class);
556    job.setOutputValueClass(MapReduceExtendedCell.class);
557    job.setOutputFormatClass(cls);
558
559    if (multiTableInfo.stream().distinct().count() != multiTableInfo.size()) {
560      throw new IllegalArgumentException("Duplicate entries found in TableInfo argument");
561    }
562    boolean writeMultipleTables = false;
563    if (MultiTableHFileOutputFormat.class.equals(cls)) {
564      writeMultipleTables = true;
565      conf.setBoolean(MULTI_TABLE_HFILEOUTPUTFORMAT_CONF_KEY, true);
566    }
567    // Based on the configured map output class, set the correct reducer to properly
568    // sort the incoming values.
569    // TODO it would be nice to pick one or the other of these formats.
570    if (KeyValue.class.equals(job.getMapOutputValueClass())
571        || MapReduceExtendedCell.class.equals(job.getMapOutputValueClass())) {
572      job.setReducerClass(CellSortReducer.class);
573    } else if (Put.class.equals(job.getMapOutputValueClass())) {
574      job.setReducerClass(PutSortReducer.class);
575    } else if (Text.class.equals(job.getMapOutputValueClass())) {
576      job.setReducerClass(TextSortReducer.class);
577    } else {
578      LOG.warn("Unknown map output value type:" + job.getMapOutputValueClass());
579    }
580
581    conf.setStrings("io.serializations", conf.get("io.serializations"),
582        MutationSerialization.class.getName(), ResultSerialization.class.getName(),
583        CellSerialization.class.getName());
584
585    if (conf.getBoolean(LOCALITY_SENSITIVE_CONF_KEY, DEFAULT_LOCALITY_SENSITIVE)) {
586      LOG.info("bulkload locality sensitive enabled");
587    }
588
589    /* Now get the region start keys for every table required */
590    List<String> allTableNames = new ArrayList<>(multiTableInfo.size());
591    List<RegionLocator> regionLocators = new ArrayList<>(multiTableInfo.size());
592    List<TableDescriptor> tableDescriptors = new ArrayList<>(multiTableInfo.size());
593
594    for(TableInfo tableInfo : multiTableInfo) {
595      regionLocators.add(tableInfo.getRegionLocator());
596      String tn = writeMultipleTables?
597        tableInfo.getRegionLocator().getName().getNameWithNamespaceInclAsString():
598        tableInfo.getRegionLocator().getName().getNameAsString();
599      allTableNames.add(tn);
600      tableDescriptors.add(tableInfo.getTableDescriptor());
601    }
602    // Record tablenames for creating writer by favored nodes, and decoding compression,
603    // block size and other attributes of columnfamily per table
604    conf.set(OUTPUT_TABLE_NAME_CONF_KEY, StringUtils.join(allTableNames, Bytes
605            .toString(tableSeparator)));
606    List<ImmutableBytesWritable> startKeys =
607      getRegionStartKeys(regionLocators, writeMultipleTables);
608    // Use table's region boundaries for TOP split points.
609    LOG.info("Configuring " + startKeys.size() + " reduce partitions " +
610        "to match current region count for all tables");
611    job.setNumReduceTasks(startKeys.size());
612
613    configurePartitioner(job, startKeys, writeMultipleTables);
614    // Set compression algorithms based on column families
615
616    conf.set(COMPRESSION_FAMILIES_CONF_KEY, serializeColumnFamilyAttribute(compressionDetails,
617            tableDescriptors));
618    conf.set(BLOCK_SIZE_FAMILIES_CONF_KEY, serializeColumnFamilyAttribute(blockSizeDetails,
619            tableDescriptors));
620    conf.set(BLOOM_TYPE_FAMILIES_CONF_KEY, serializeColumnFamilyAttribute(bloomTypeDetails,
621            tableDescriptors));
622    conf.set(BLOOM_PARAM_FAMILIES_CONF_KEY, serializeColumnFamilyAttribute(bloomParamDetails,
623        tableDescriptors));
624    conf.set(DATABLOCK_ENCODING_FAMILIES_CONF_KEY,
625            serializeColumnFamilyAttribute(dataBlockEncodingDetails, tableDescriptors));
626
627    TableMapReduceUtil.addDependencyJars(job);
628    TableMapReduceUtil.initCredentials(job);
629    LOG.info("Incremental output configured for tables: " + StringUtils.join(allTableNames, ","));
630  }
631
632  public static void configureIncrementalLoadMap(Job job, TableDescriptor tableDescriptor) throws
633      IOException {
634    Configuration conf = job.getConfiguration();
635
636    job.setOutputKeyClass(ImmutableBytesWritable.class);
637    job.setOutputValueClass(MapReduceExtendedCell.class);
638    job.setOutputFormatClass(HFileOutputFormat2.class);
639
640    ArrayList<TableDescriptor> singleTableDescriptor = new ArrayList<>(1);
641    singleTableDescriptor.add(tableDescriptor);
642
643    conf.set(OUTPUT_TABLE_NAME_CONF_KEY, tableDescriptor.getTableName().getNameAsString());
644    // Set compression algorithms based on column families
645    conf.set(COMPRESSION_FAMILIES_CONF_KEY,
646        serializeColumnFamilyAttribute(compressionDetails, singleTableDescriptor));
647    conf.set(BLOCK_SIZE_FAMILIES_CONF_KEY,
648        serializeColumnFamilyAttribute(blockSizeDetails, singleTableDescriptor));
649    conf.set(BLOOM_TYPE_FAMILIES_CONF_KEY,
650        serializeColumnFamilyAttribute(bloomTypeDetails, singleTableDescriptor));
651    conf.set(BLOOM_PARAM_FAMILIES_CONF_KEY,
652        serializeColumnFamilyAttribute(bloomParamDetails, singleTableDescriptor));
653    conf.set(DATABLOCK_ENCODING_FAMILIES_CONF_KEY,
654        serializeColumnFamilyAttribute(dataBlockEncodingDetails, singleTableDescriptor));
655
656    TableMapReduceUtil.addDependencyJars(job);
657    TableMapReduceUtil.initCredentials(job);
658    LOG.info("Incremental table " + tableDescriptor.getTableName() + " output configured.");
659  }
660
661  /**
662   * Runs inside the task to deserialize column family to compression algorithm
663   * map from the configuration.
664   *
665   * @param conf to read the serialized values from
666   * @return a map from column family to the configured compression algorithm
667   */
668  @VisibleForTesting
669  static Map<byte[], Algorithm> createFamilyCompressionMap(Configuration
670      conf) {
671    Map<byte[], String> stringMap = createFamilyConfValueMap(conf,
672        COMPRESSION_FAMILIES_CONF_KEY);
673    Map<byte[], Algorithm> compressionMap = new TreeMap<>(Bytes.BYTES_COMPARATOR);
674    for (Map.Entry<byte[], String> e : stringMap.entrySet()) {
675      Algorithm algorithm = HFileWriterImpl.compressionByName(e.getValue());
676      compressionMap.put(e.getKey(), algorithm);
677    }
678    return compressionMap;
679  }
680
681  /**
682   * Runs inside the task to deserialize column family to bloom filter type
683   * map from the configuration.
684   *
685   * @param conf to read the serialized values from
686   * @return a map from column family to the the configured bloom filter type
687   */
688  @VisibleForTesting
689  static Map<byte[], BloomType> createFamilyBloomTypeMap(Configuration conf) {
690    Map<byte[], String> stringMap = createFamilyConfValueMap(conf,
691        BLOOM_TYPE_FAMILIES_CONF_KEY);
692    Map<byte[], BloomType> bloomTypeMap = new TreeMap<>(Bytes.BYTES_COMPARATOR);
693    for (Map.Entry<byte[], String> e : stringMap.entrySet()) {
694      BloomType bloomType = BloomType.valueOf(e.getValue());
695      bloomTypeMap.put(e.getKey(), bloomType);
696    }
697    return bloomTypeMap;
698  }
699
700  /**
701   * Runs inside the task to deserialize column family to bloom filter param
702   * map from the configuration.
703   *
704   * @param conf to read the serialized values from
705   * @return a map from column family to the the configured bloom filter param
706   */
707  @VisibleForTesting
708  static Map<byte[], String> createFamilyBloomParamMap(Configuration conf) {
709    return createFamilyConfValueMap(conf, BLOOM_PARAM_FAMILIES_CONF_KEY);
710  }
711
712
713  /**
714   * Runs inside the task to deserialize column family to block size
715   * map from the configuration.
716   *
717   * @param conf to read the serialized values from
718   * @return a map from column family to the configured block size
719   */
720  @VisibleForTesting
721  static Map<byte[], Integer> createFamilyBlockSizeMap(Configuration conf) {
722    Map<byte[], String> stringMap = createFamilyConfValueMap(conf,
723        BLOCK_SIZE_FAMILIES_CONF_KEY);
724    Map<byte[], Integer> blockSizeMap = new TreeMap<>(Bytes.BYTES_COMPARATOR);
725    for (Map.Entry<byte[], String> e : stringMap.entrySet()) {
726      Integer blockSize = Integer.parseInt(e.getValue());
727      blockSizeMap.put(e.getKey(), blockSize);
728    }
729    return blockSizeMap;
730  }
731
732  /**
733   * Runs inside the task to deserialize column family to data block encoding
734   * type map from the configuration.
735   *
736   * @param conf to read the serialized values from
737   * @return a map from column family to HFileDataBlockEncoder for the
738   *         configured data block type for the family
739   */
740  @VisibleForTesting
741  static Map<byte[], DataBlockEncoding> createFamilyDataBlockEncodingMap(
742      Configuration conf) {
743    Map<byte[], String> stringMap = createFamilyConfValueMap(conf,
744        DATABLOCK_ENCODING_FAMILIES_CONF_KEY);
745    Map<byte[], DataBlockEncoding> encoderMap = new TreeMap<>(Bytes.BYTES_COMPARATOR);
746    for (Map.Entry<byte[], String> e : stringMap.entrySet()) {
747      encoderMap.put(e.getKey(), DataBlockEncoding.valueOf((e.getValue())));
748    }
749    return encoderMap;
750  }
751
752
753  /**
754   * Run inside the task to deserialize column family to given conf value map.
755   *
756   * @param conf to read the serialized values from
757   * @param confName conf key to read from the configuration
758   * @return a map of column family to the given configuration value
759   */
760  private static Map<byte[], String> createFamilyConfValueMap(
761      Configuration conf, String confName) {
762    Map<byte[], String> confValMap = new TreeMap<>(Bytes.BYTES_COMPARATOR);
763    String confVal = conf.get(confName, "");
764    for (String familyConf : confVal.split("&")) {
765      String[] familySplit = familyConf.split("=");
766      if (familySplit.length != 2) {
767        continue;
768      }
769      try {
770        confValMap.put(Bytes.toBytes(URLDecoder.decode(familySplit[0], "UTF-8")),
771            URLDecoder.decode(familySplit[1], "UTF-8"));
772      } catch (UnsupportedEncodingException e) {
773        // will not happen with UTF-8 encoding
774        throw new AssertionError(e);
775      }
776    }
777    return confValMap;
778  }
779
780  /**
781   * Configure <code>job</code> with a TotalOrderPartitioner, partitioning against
782   * <code>splitPoints</code>. Cleans up the partitions file after job exists.
783   */
784  static void configurePartitioner(Job job, List<ImmutableBytesWritable> splitPoints, boolean
785          writeMultipleTables)
786      throws IOException {
787    Configuration conf = job.getConfiguration();
788    // create the partitions file
789    FileSystem fs = FileSystem.get(conf);
790    String hbaseTmpFsDir =
791        conf.get(HConstants.TEMPORARY_FS_DIRECTORY_KEY,
792            fs.getHomeDirectory() + "/hbase-staging");
793    Path partitionsPath = new Path(hbaseTmpFsDir, "partitions_" + UUID.randomUUID());
794    fs.makeQualified(partitionsPath);
795    writePartitions(conf, partitionsPath, splitPoints, writeMultipleTables);
796    fs.deleteOnExit(partitionsPath);
797
798    // configure job to use it
799    job.setPartitionerClass(TotalOrderPartitioner.class);
800    TotalOrderPartitioner.setPartitionFile(conf, partitionsPath);
801  }
802
803  @edu.umd.cs.findbugs.annotations.SuppressWarnings(value =
804    "RCN_REDUNDANT_NULLCHECK_OF_NONNULL_VALUE")
805  @VisibleForTesting
806  static String serializeColumnFamilyAttribute(Function<ColumnFamilyDescriptor, String> fn,
807        List<TableDescriptor> allTables)
808      throws UnsupportedEncodingException {
809    StringBuilder attributeValue = new StringBuilder();
810    int i = 0;
811    for (TableDescriptor tableDescriptor : allTables) {
812      if (tableDescriptor == null) {
813        // could happen with mock table instance
814        // CODEREVIEW: Can I set an empty string in conf if mock table instance?
815        return "";
816      }
817      for (ColumnFamilyDescriptor familyDescriptor : tableDescriptor.getColumnFamilies()) {
818        if (i++ > 0) {
819          attributeValue.append('&');
820        }
821        attributeValue.append(URLEncoder.encode(
822          Bytes.toString(combineTableNameSuffix(tableDescriptor.getTableName().getName(),
823            familyDescriptor.getName())), "UTF-8"));
824        attributeValue.append('=');
825        attributeValue.append(URLEncoder.encode(fn.apply(familyDescriptor), "UTF-8"));
826      }
827    }
828    // Get rid of the last ampersand
829    return attributeValue.toString();
830  }
831
832  /**
833   * Serialize column family to compression algorithm map to configuration.
834   * Invoked while configuring the MR job for incremental load.
835   */
836  @VisibleForTesting
837  static Function<ColumnFamilyDescriptor, String> compressionDetails = familyDescriptor ->
838          familyDescriptor.getCompressionType().getName();
839
840  /**
841   * Serialize column family to block size map to configuration. Invoked while
842   * configuring the MR job for incremental load.
843   */
844  @VisibleForTesting
845  static Function<ColumnFamilyDescriptor, String> blockSizeDetails = familyDescriptor -> String
846          .valueOf(familyDescriptor.getBlocksize());
847
848  /**
849   * Serialize column family to bloom type map to configuration. Invoked while
850   * configuring the MR job for incremental load.
851   */
852  @VisibleForTesting
853  static Function<ColumnFamilyDescriptor, String> bloomTypeDetails = familyDescriptor -> {
854    String bloomType = familyDescriptor.getBloomFilterType().toString();
855    if (bloomType == null) {
856      bloomType = ColumnFamilyDescriptorBuilder.DEFAULT_BLOOMFILTER.name();
857    }
858    return bloomType;
859  };
860
861  /**
862   * Serialize column family to bloom param map to configuration. Invoked while
863   * configuring the MR job for incremental load.
864   */
865  @VisibleForTesting
866  static Function<ColumnFamilyDescriptor, String> bloomParamDetails = familyDescriptor -> {
867    BloomType bloomType = familyDescriptor.getBloomFilterType();
868    String bloomParam = "";
869    if (bloomType == BloomType.ROWPREFIX_FIXED_LENGTH) {
870      bloomParam = familyDescriptor.getConfigurationValue(BloomFilterUtil.PREFIX_LENGTH_KEY);
871    }
872    return bloomParam;
873  };
874
875  /**
876   * Serialize column family to data block encoding map to configuration.
877   * Invoked while configuring the MR job for incremental load.
878   */
879  @VisibleForTesting
880  static Function<ColumnFamilyDescriptor, String> dataBlockEncodingDetails = familyDescriptor -> {
881    DataBlockEncoding encoding = familyDescriptor.getDataBlockEncoding();
882    if (encoding == null) {
883      encoding = DataBlockEncoding.NONE;
884    }
885    return encoding.toString();
886  };
887
888}