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