1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.mob.filecompactions;
20
21 import java.io.FileNotFoundException;
22 import java.io.IOException;
23 import java.util.ArrayList;
24 import java.util.Collection;
25 import java.util.Collections;
26 import java.util.Date;
27 import java.util.HashMap;
28 import java.util.List;
29 import java.util.Map;
30 import java.util.Map.Entry;
31 import java.util.concurrent.Callable;
32 import java.util.concurrent.ExecutorService;
33 import java.util.concurrent.Future;
34
35 import org.apache.commons.logging.Log;
36 import org.apache.commons.logging.LogFactory;
37 import org.apache.hadoop.classification.InterfaceAudience;
38 import org.apache.hadoop.conf.Configuration;
39 import org.apache.hadoop.fs.FileStatus;
40 import org.apache.hadoop.fs.FileSystem;
41 import org.apache.hadoop.fs.Path;
42 import org.apache.hadoop.hbase.Cell;
43 import org.apache.hadoop.hbase.HColumnDescriptor;
44 import org.apache.hadoop.hbase.HConstants;
45 import org.apache.hadoop.hbase.KeyValue;
46 import org.apache.hadoop.hbase.KeyValueUtil;
47 import org.apache.hadoop.hbase.TableName;
48 import org.apache.hadoop.hbase.Tag;
49 import org.apache.hadoop.hbase.TagType;
50 import org.apache.hadoop.hbase.client.HTable;
51 import org.apache.hadoop.hbase.client.Scan;
52 import org.apache.hadoop.hbase.io.HFileLink;
53 import org.apache.hadoop.hbase.io.hfile.CacheConfig;
54 import org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles;
55 import org.apache.hadoop.hbase.mob.MobConstants;
56 import org.apache.hadoop.hbase.mob.MobFileName;
57 import org.apache.hadoop.hbase.mob.MobUtils;
58 import org.apache.hadoop.hbase.mob.filecompactions.MobFileCompactionRequest.CompactionType;
59 import org.apache.hadoop.hbase.mob.filecompactions.PartitionedMobFileCompactionRequest.CompactionPartition;
60 import org.apache.hadoop.hbase.mob.filecompactions.PartitionedMobFileCompactionRequest.CompactionPartitionId;
61 import org.apache.hadoop.hbase.regionserver.*;
62 import org.apache.hadoop.hbase.regionserver.StoreFile.Writer;
63 import org.apache.hadoop.hbase.regionserver.ScannerContext;
64 import org.apache.hadoop.hbase.util.Bytes;
65 import org.apache.hadoop.hbase.util.Pair;
66
67
68
69
70 @InterfaceAudience.Private
71 public class PartitionedMobFileCompactor extends MobFileCompactor {
72
73 private static final Log LOG = LogFactory.getLog(PartitionedMobFileCompactor.class);
74 protected long mergeableSize;
75 protected int delFileMaxCount;
76
77 protected int compactionBatchSize;
78 protected int compactionKVMax;
79
80 private Path tempPath;
81 private Path bulkloadPath;
82 private CacheConfig compactionCacheConfig;
83 private Tag tableNameTag;
84
85 public PartitionedMobFileCompactor(Configuration conf, FileSystem fs, TableName tableName,
86 HColumnDescriptor column, ExecutorService pool) {
87 super(conf, fs, tableName, column, pool);
88 mergeableSize = conf.getLong(MobConstants.MOB_FILE_COMPACTION_MERGEABLE_THRESHOLD,
89 MobConstants.DEFAULT_MOB_FILE_COMPACTION_MERGEABLE_THRESHOLD);
90 delFileMaxCount = conf.getInt(MobConstants.MOB_DELFILE_MAX_COUNT,
91 MobConstants.DEFAULT_MOB_DELFILE_MAX_COUNT);
92
93 compactionBatchSize = conf.getInt(MobConstants.MOB_FILE_COMPACTION_BATCH_SIZE,
94 MobConstants.DEFAULT_MOB_FILE_COMPACTION_BATCH_SIZE);
95 tempPath = new Path(MobUtils.getMobHome(conf), MobConstants.TEMP_DIR_NAME);
96 bulkloadPath = new Path(tempPath, new Path(MobConstants.BULKLOAD_DIR_NAME, new Path(
97 tableName.getNamespaceAsString(), tableName.getQualifierAsString())));
98 compactionKVMax = this.conf.getInt(HConstants.COMPACTION_KV_MAX,
99 HConstants.COMPACTION_KV_MAX_DEFAULT);
100 Configuration copyOfConf = new Configuration(conf);
101 copyOfConf.setFloat(HConstants.HFILE_BLOCK_CACHE_SIZE_KEY, 0f);
102 compactionCacheConfig = new CacheConfig(copyOfConf);
103 tableNameTag = new Tag(TagType.MOB_TABLE_NAME_TAG_TYPE, tableName.getName());
104 }
105
106 @Override
107 public List<Path> compact(List<FileStatus> files, boolean isForceAllFiles) throws IOException {
108 if (files == null || files.isEmpty()) {
109 LOG.info("No candidate mob files");
110 return null;
111 }
112 LOG.info("isForceAllFiles: " + isForceAllFiles);
113
114 PartitionedMobFileCompactionRequest request = select(files, isForceAllFiles);
115
116 return performCompaction(request);
117 }
118
119
120
121
122
123
124
125
126
127 protected PartitionedMobFileCompactionRequest select(List<FileStatus> candidates,
128 boolean isForceAllFiles) throws IOException {
129 Collection<FileStatus> allDelFiles = new ArrayList<FileStatus>();
130 Map<CompactionPartitionId, CompactionPartition> filesToCompact =
131 new HashMap<CompactionPartitionId, CompactionPartition>();
132 int selectedFileCount = 0;
133 int irrelevantFileCount = 0;
134 for (FileStatus file : candidates) {
135 if (!file.isFile()) {
136 irrelevantFileCount++;
137 continue;
138 }
139
140 FileStatus linkedFile = file;
141 if (HFileLink.isHFileLink(file.getPath())) {
142 HFileLink link = HFileLink.buildFromHFileLinkPattern(conf, file.getPath());
143 linkedFile = getLinkedFileStatus(link);
144 if (linkedFile == null) {
145
146 irrelevantFileCount++;
147 continue;
148 }
149 }
150 if (StoreFileInfo.isDelFile(linkedFile.getPath())) {
151 allDelFiles.add(file);
152 } else if (isForceAllFiles || linkedFile.getLen() < mergeableSize) {
153
154
155 MobFileName fileName = MobFileName.create(linkedFile.getPath().getName());
156 CompactionPartitionId id = new CompactionPartitionId(fileName.getStartKey(),
157 fileName.getDate());
158 CompactionPartition compactionPartition = filesToCompact.get(id);
159 if (compactionPartition == null) {
160 compactionPartition = new CompactionPartition(id);
161 compactionPartition.addFile(file);
162 filesToCompact.put(id, compactionPartition);
163 } else {
164 compactionPartition.addFile(file);
165 }
166 selectedFileCount++;
167 }
168 }
169 PartitionedMobFileCompactionRequest request = new PartitionedMobFileCompactionRequest(
170 filesToCompact.values(), allDelFiles);
171 if (candidates.size() == (allDelFiles.size() + selectedFileCount + irrelevantFileCount)) {
172
173 request.setCompactionType(CompactionType.ALL_FILES);
174 }
175 LOG.info("The compaction type is " + request.getCompactionType() + ", the request has "
176 + allDelFiles.size() + " del files, " + selectedFileCount + " selected files, and "
177 + irrelevantFileCount + " irrelevant files");
178 return request;
179 }
180
181
182
183
184
185
186
187
188
189
190
191
192 protected List<Path> performCompaction(PartitionedMobFileCompactionRequest request)
193 throws IOException {
194
195 List<Path> delFilePaths = new ArrayList<Path>();
196 for (FileStatus delFile : request.delFiles) {
197 delFilePaths.add(delFile.getPath());
198 }
199 List<Path> newDelPaths = compactDelFiles(request, delFilePaths);
200 List<StoreFile> newDelFiles = new ArrayList<StoreFile>();
201 List<Path> paths = null;
202 try {
203 for (Path newDelPath : newDelPaths) {
204 StoreFile sf = new StoreFile(fs, newDelPath, conf, compactionCacheConfig, BloomType.NONE);
205
206
207 sf.createReader();
208 newDelFiles.add(sf);
209 }
210 LOG.info("After merging, there are " + newDelFiles.size() + " del files");
211
212 paths = compactMobFiles(request, newDelFiles);
213 LOG.info("After compaction, there are " + paths.size() + " mob files");
214 } finally {
215 closeStoreFileReaders(newDelFiles);
216 }
217
218 if (request.type == CompactionType.ALL_FILES && !newDelPaths.isEmpty()) {
219 LOG.info("After a mob file compaction with all files selected, archiving the del files "
220 + newDelFiles);
221 try {
222 MobUtils.removeMobFiles(conf, fs, tableName, mobTableDir, column.getName(), newDelFiles);
223 } catch (IOException e) {
224 LOG.error("Failed to archive the del files " + newDelFiles, e);
225 }
226 }
227 return paths;
228 }
229
230
231
232
233
234
235
236
237 protected List<Path> compactMobFiles(final PartitionedMobFileCompactionRequest request,
238 final List<StoreFile> delFiles) throws IOException {
239 Collection<CompactionPartition> partitions = request.compactionPartitions;
240 if (partitions == null || partitions.isEmpty()) {
241 LOG.info("No partitions of mob files");
242 return Collections.emptyList();
243 }
244 List<Path> paths = new ArrayList<Path>();
245 final HTable table = new HTable(conf, tableName);
246 try {
247 Map<CompactionPartitionId, Future<List<Path>>> results =
248 new HashMap<CompactionPartitionId, Future<List<Path>>>();
249
250 for (final CompactionPartition partition : partitions) {
251 results.put(partition.getPartitionId(), pool.submit(new Callable<List<Path>>() {
252 @Override
253 public List<Path> call() throws Exception {
254 LOG.info("Compacting mob files for partition " + partition.getPartitionId());
255 return compactMobFilePartition(request, partition, delFiles, table);
256 }
257 }));
258 }
259
260 boolean hasFailure = false;
261 for (Entry<CompactionPartitionId, Future<List<Path>>> result : results.entrySet()) {
262 try {
263 paths.addAll(result.getValue().get());
264 } catch (Exception e) {
265
266 LOG.error("Failed to compact the partition " + result.getKey(), e);
267 hasFailure = true;
268 }
269 }
270 if (hasFailure) {
271
272 throw new IOException("Failed to compact the partitions");
273 }
274 } finally {
275 try {
276 table.close();
277 } catch (IOException e) {
278 LOG.error("Failed to close the HTable", e);
279 }
280 }
281 return paths;
282 }
283
284
285
286
287
288
289
290
291
292
293 private List<Path> compactMobFilePartition(PartitionedMobFileCompactionRequest request,
294 CompactionPartition partition, List<StoreFile> delFiles, HTable table) throws IOException {
295 List<Path> newFiles = new ArrayList<Path>();
296 List<FileStatus> files = partition.listFiles();
297 int offset = 0;
298 Path bulkloadPathOfPartition = new Path(bulkloadPath, partition.getPartitionId().toString());
299 Path bulkloadColumnPath = new Path(bulkloadPathOfPartition, column.getNameAsString());
300 while (offset < files.size()) {
301 int batch = compactionBatchSize;
302 if (files.size() - offset < compactionBatchSize) {
303 batch = files.size() - offset;
304 }
305 if (batch == 1 && delFiles.isEmpty()) {
306
307
308 newFiles.add(files.get(offset).getPath());
309 offset++;
310 continue;
311 }
312
313 fs.delete(bulkloadPathOfPartition, true);
314
315 List<StoreFile> filesToCompact = new ArrayList<StoreFile>();
316 for (int i = offset; i < batch + offset; i++) {
317 StoreFile sf = new StoreFile(fs, files.get(i).getPath(), conf, compactionCacheConfig,
318 BloomType.NONE);
319 filesToCompact.add(sf);
320 }
321 filesToCompact.addAll(delFiles);
322
323 compactMobFilesInBatch(request, partition, table, filesToCompact, batch,
324 bulkloadPathOfPartition, bulkloadColumnPath, newFiles);
325
326 offset += batch;
327 }
328 LOG.info("Compaction is finished. The number of mob files is changed from " + files.size()
329 + " to " + newFiles.size());
330 return newFiles;
331 }
332
333
334
335
336
337 private void closeStoreFileReaders(List<StoreFile> storeFiles) {
338 for (StoreFile storeFile : storeFiles) {
339 try {
340 storeFile.closeReader(true);
341 } catch (IOException e) {
342 LOG.warn("Failed to close the reader on store file " + storeFile.getPath(), e);
343 }
344 }
345 }
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361 private void compactMobFilesInBatch(PartitionedMobFileCompactionRequest request,
362 CompactionPartition partition, HTable table, List<StoreFile> filesToCompact, int batch,
363 Path bulkloadPathOfPartition, Path bulkloadColumnPath, List<Path> newFiles)
364 throws IOException {
365
366 StoreScanner scanner = createScanner(filesToCompact, ScanType.COMPACT_DROP_DELETES);
367
368 List<StoreFile> mobFilesToCompact = filesToCompact.subList(0, batch);
369
370 Pair<Long, Long> fileInfo = getFileInfo(mobFilesToCompact);
371
372 Writer writer = null;
373 Writer refFileWriter = null;
374 Path filePath = null;
375 Path refFilePath = null;
376 long mobCells = 0;
377 try {
378 writer = MobUtils.createWriter(conf, fs, column, partition.getPartitionId().getDate(),
379 tempPath, Long.MAX_VALUE, column.getCompactionCompression(), partition.getPartitionId()
380 .getStartKey(), compactionCacheConfig);
381 filePath = writer.getPath();
382 byte[] fileName = Bytes.toBytes(filePath.getName());
383
384 refFileWriter = MobUtils.createRefFileWriter(conf, fs, column, bulkloadColumnPath, fileInfo
385 .getSecond().longValue(), compactionCacheConfig);
386 refFilePath = refFileWriter.getPath();
387 List<Cell> cells = new ArrayList<Cell>();
388 boolean hasMore = false;
389
390 ScannerContext scannerContext =
391 ScannerContext.newBuilder().setBatchLimit(compactionKVMax).build();
392 do {
393 hasMore = scanner.next(cells, scannerContext);
394 for (Cell cell : cells) {
395
396 KeyValue kv = KeyValueUtil.ensureKeyValue(cell);
397
398 writer.append(kv);
399
400 KeyValue reference = MobUtils.createMobRefKeyValue(kv, fileName, tableNameTag);
401 refFileWriter.append(reference);
402 mobCells++;
403 }
404 cells.clear();
405 } while (hasMore);
406 } finally {
407
408 scanner.close();
409
410 closeMobFileWriter(writer, fileInfo.getFirst(), mobCells);
411
412 closeRefFileWriter(refFileWriter, fileInfo.getFirst(), request.selectionTime);
413 }
414 if (mobCells > 0) {
415
416 MobUtils.commitFile(conf, fs, filePath, mobFamilyDir, compactionCacheConfig);
417
418 bulkloadRefFile(table, bulkloadPathOfPartition, filePath.getName());
419 newFiles.add(new Path(mobFamilyDir, filePath.getName()));
420 } else {
421
422
423 deletePath(filePath);
424
425 deletePath(refFilePath);
426 }
427
428 try {
429 closeStoreFileReaders(mobFilesToCompact);
430 MobUtils
431 .removeMobFiles(conf, fs, tableName, mobTableDir, column.getName(), mobFilesToCompact);
432 } catch (IOException e) {
433 LOG.error("Failed to archive the files " + mobFilesToCompact, e);
434 }
435 }
436
437
438
439
440
441
442
443
444
445 protected List<Path> compactDelFiles(PartitionedMobFileCompactionRequest request,
446 List<Path> delFilePaths) throws IOException {
447 if (delFilePaths.size() <= delFileMaxCount) {
448 return delFilePaths;
449 }
450
451 int offset = 0;
452 List<Path> paths = new ArrayList<Path>();
453 while (offset < delFilePaths.size()) {
454
455 int batch = compactionBatchSize;
456 if (delFilePaths.size() - offset < compactionBatchSize) {
457 batch = delFilePaths.size() - offset;
458 }
459 List<StoreFile> batchedDelFiles = new ArrayList<StoreFile>();
460 if (batch == 1) {
461
462 paths.add(delFilePaths.get(offset));
463 offset++;
464 continue;
465 }
466 for (int i = offset; i < batch + offset; i++) {
467 batchedDelFiles.add(new StoreFile(fs, delFilePaths.get(i), conf, compactionCacheConfig,
468 BloomType.NONE));
469 }
470
471 paths.add(compactDelFilesInBatch(request, batchedDelFiles));
472
473 offset += batch;
474 }
475 return compactDelFiles(request, paths);
476 }
477
478
479
480
481
482
483
484
485 private Path compactDelFilesInBatch(PartitionedMobFileCompactionRequest request,
486 List<StoreFile> delFiles) throws IOException {
487
488 StoreScanner scanner = createScanner(delFiles, ScanType.COMPACT_RETAIN_DELETES);
489 Writer writer = null;
490 Path filePath = null;
491 try {
492 writer = MobUtils.createDelFileWriter(conf, fs, column,
493 MobUtils.formatDate(new Date(request.selectionTime)), tempPath, Long.MAX_VALUE,
494 column.getCompactionCompression(), HConstants.EMPTY_START_ROW, compactionCacheConfig);
495 filePath = writer.getPath();
496 List<Cell> cells = new ArrayList<Cell>();
497 boolean hasMore = false;
498 ScannerContext scannerContext =
499 ScannerContext.newBuilder().setBatchLimit(compactionKVMax).build();
500 do {
501 hasMore = scanner.next(cells, scannerContext);
502 for (Cell cell : cells) {
503
504 KeyValue kv = KeyValueUtil.ensureKeyValue(cell);
505 writer.append(kv);
506 }
507 cells.clear();
508 } while (hasMore);
509 } finally {
510 scanner.close();
511 if (writer != null) {
512 try {
513 writer.close();
514 } catch (IOException e) {
515 LOG.error("Failed to close the writer of the file " + filePath, e);
516 }
517 }
518 }
519
520 Path path = MobUtils.commitFile(conf, fs, filePath, mobFamilyDir, compactionCacheConfig);
521
522 try {
523 MobUtils.removeMobFiles(conf, fs, tableName, mobTableDir, column.getName(), delFiles);
524 } catch (IOException e) {
525 LOG.error("Failed to archive the old del files " + delFiles, e);
526 }
527 return path;
528 }
529
530
531
532
533
534
535
536
537 private StoreScanner createScanner(List<StoreFile> filesToCompact, ScanType scanType)
538 throws IOException {
539 List scanners = StoreFileScanner.getScannersForStoreFiles(filesToCompact, false, true, false,
540 false, HConstants.LATEST_TIMESTAMP);
541 Scan scan = new Scan();
542 scan.setMaxVersions(column.getMaxVersions());
543 long ttl = HStore.determineTTLFromFamily(column);
544 ScanInfo scanInfo = new ScanInfo(conf, column, ttl, 0, KeyValue.COMPARATOR);
545 StoreScanner scanner = new StoreScanner(scan, scanInfo, scanType, null, scanners, 0L,
546 HConstants.LATEST_TIMESTAMP);
547 return scanner;
548 }
549
550
551
552
553
554
555
556
557 private void bulkloadRefFile(HTable table, Path bulkloadDirectory, String fileName)
558 throws IOException {
559
560 try {
561 LoadIncrementalHFiles bulkload = new LoadIncrementalHFiles(conf);
562 bulkload.doBulkLoad(bulkloadDirectory, table);
563 } catch (Exception e) {
564
565 deletePath(new Path(mobFamilyDir, fileName));
566 throw new IOException(e);
567 } finally {
568
569 deletePath(bulkloadDirectory);
570 }
571 }
572
573
574
575
576
577
578
579
580 private void closeMobFileWriter(Writer writer, long maxSeqId, long mobCellsCount)
581 throws IOException {
582 if (writer != null) {
583 writer.appendMetadata(maxSeqId, false, mobCellsCount);
584 try {
585 writer.close();
586 } catch (IOException e) {
587 LOG.error("Failed to close the writer of the file " + writer.getPath(), e);
588 }
589 }
590 }
591
592
593
594
595
596
597
598
599 private void closeRefFileWriter(Writer writer, long maxSeqId, long bulkloadTime)
600 throws IOException {
601 if (writer != null) {
602 writer.appendMetadata(maxSeqId, false);
603 writer.appendFileInfo(StoreFile.BULKLOAD_TIME_KEY, Bytes.toBytes(bulkloadTime));
604 try {
605 writer.close();
606 } catch (IOException e) {
607 LOG.error("Failed to close the writer of the ref file " + writer.getPath(), e);
608 }
609 }
610 }
611
612
613
614
615
616
617
618 private Pair<Long, Long> getFileInfo(List<StoreFile> storeFiles) throws IOException {
619 long maxSeqId = 0;
620 long maxKeyCount = 0;
621 for (StoreFile sf : storeFiles) {
622
623 maxSeqId = Math.max(maxSeqId, sf.getMaxSequenceId());
624 byte[] count = sf.createReader().loadFileInfo().get(StoreFile.MOB_CELLS_COUNT);
625 if (count != null) {
626 maxKeyCount += Bytes.toLong(count);
627 }
628 }
629 return new Pair<Long, Long>(Long.valueOf(maxSeqId), Long.valueOf(maxKeyCount));
630 }
631
632
633
634
635
636 private void deletePath(Path path) {
637 try {
638 if (path != null) {
639 fs.delete(path, true);
640 }
641 } catch (IOException e) {
642 LOG.error("Failed to delete the file " + path, e);
643 }
644 }
645
646 private FileStatus getLinkedFileStatus(HFileLink link) throws IOException {
647 Path[] locations = link.getLocations();
648 for (Path location : locations) {
649 FileStatus file = getFileStatus(location);
650 if (file != null) {
651 return file;
652 }
653 }
654 return null;
655 }
656
657 private FileStatus getFileStatus(Path path) throws IOException {
658 try {
659 if (path != null) {
660 FileStatus file = fs.getFileStatus(path);
661 return file;
662 }
663 } catch (FileNotFoundException e) {
664 LOG.warn("The file " + path + " can not be found", e);
665 }
666 return null;
667 }
668 }