1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.snapshot;
20
21 import java.io.BufferedInputStream;
22 import java.io.FileNotFoundException;
23 import java.io.DataInput;
24 import java.io.DataOutput;
25 import java.io.IOException;
26 import java.io.InputStream;
27 import java.util.ArrayList;
28 import java.util.Collections;
29 import java.util.Comparator;
30 import java.util.LinkedList;
31 import java.util.List;
32 import java.util.Random;
33
34 import org.apache.commons.logging.Log;
35 import org.apache.commons.logging.LogFactory;
36 import org.apache.hadoop.hbase.classification.InterfaceAudience;
37 import org.apache.hadoop.hbase.classification.InterfaceStability;
38 import org.apache.hadoop.conf.Configuration;
39 import org.apache.hadoop.conf.Configured;
40 import org.apache.hadoop.fs.FSDataInputStream;
41 import org.apache.hadoop.fs.FSDataOutputStream;
42 import org.apache.hadoop.fs.FileChecksum;
43 import org.apache.hadoop.fs.FileStatus;
44 import org.apache.hadoop.fs.FileSystem;
45 import org.apache.hadoop.fs.FileUtil;
46 import org.apache.hadoop.fs.Path;
47 import org.apache.hadoop.fs.permission.FsPermission;
48 import org.apache.hadoop.hbase.TableName;
49 import org.apache.hadoop.hbase.HBaseConfiguration;
50 import org.apache.hadoop.hbase.HConstants;
51 import org.apache.hadoop.hbase.HRegionInfo;
52 import org.apache.hadoop.hbase.io.FileLink;
53 import org.apache.hadoop.hbase.io.HFileLink;
54 import org.apache.hadoop.hbase.io.WALLink;
55 import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
56 import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.SnapshotDescription;
57 import org.apache.hadoop.hbase.protobuf.generated.SnapshotProtos.SnapshotFileInfo;
58 import org.apache.hadoop.hbase.protobuf.generated.SnapshotProtos.SnapshotRegionManifest;
59 import org.apache.hadoop.hbase.util.FSUtils;
60 import org.apache.hadoop.hbase.util.Pair;
61 import org.apache.hadoop.io.BytesWritable;
62 import org.apache.hadoop.io.IOUtils;
63 import org.apache.hadoop.io.NullWritable;
64 import org.apache.hadoop.io.Writable;
65 import org.apache.hadoop.mapreduce.Job;
66 import org.apache.hadoop.mapreduce.JobContext;
67 import org.apache.hadoop.mapreduce.Mapper;
68 import org.apache.hadoop.mapreduce.InputFormat;
69 import org.apache.hadoop.mapreduce.InputSplit;
70 import org.apache.hadoop.mapreduce.RecordReader;
71 import org.apache.hadoop.mapreduce.TaskAttemptContext;
72 import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;
73 import org.apache.hadoop.mapreduce.security.TokenCache;
74 import org.apache.hadoop.hbase.io.hadoopbackport.ThrottledInputStream;
75 import org.apache.hadoop.util.StringUtils;
76 import org.apache.hadoop.util.Tool;
77 import org.apache.hadoop.util.ToolRunner;
78
79
80
81
82
83
84
85
86 @InterfaceAudience.Public
87 @InterfaceStability.Evolving
88 public class ExportSnapshot extends Configured implements Tool {
89 public static final String NAME = "exportsnapshot";
90
91 public static final String CONF_SOURCE_PREFIX = NAME + ".from.";
92
93 public static final String CONF_DEST_PREFIX = NAME + ".to.";
94
95 private static final Log LOG = LogFactory.getLog(ExportSnapshot.class);
96
97 private static final String MR_NUM_MAPS = "mapreduce.job.maps";
98 private static final String CONF_NUM_SPLITS = "snapshot.export.format.splits";
99 private static final String CONF_SNAPSHOT_NAME = "snapshot.export.format.snapshot.name";
100 private static final String CONF_SNAPSHOT_DIR = "snapshot.export.format.snapshot.dir";
101 private static final String CONF_FILES_USER = "snapshot.export.files.attributes.user";
102 private static final String CONF_FILES_GROUP = "snapshot.export.files.attributes.group";
103 private static final String CONF_FILES_MODE = "snapshot.export.files.attributes.mode";
104 private static final String CONF_CHECKSUM_VERIFY = "snapshot.export.checksum.verify";
105 private static final String CONF_OUTPUT_ROOT = "snapshot.export.output.root";
106 private static final String CONF_INPUT_ROOT = "snapshot.export.input.root";
107 private static final String CONF_BUFFER_SIZE = "snapshot.export.buffer.size";
108 private static final String CONF_MAP_GROUP = "snapshot.export.default.map.group";
109 private static final String CONF_BANDWIDTH_MB = "snapshot.export.map.bandwidth.mb";
110 protected static final String CONF_SKIP_TMP = "snapshot.export.skip.tmp";
111
112 static final String CONF_TEST_FAILURE = "test.snapshot.export.failure";
113 static final String CONF_TEST_RETRY = "test.snapshot.export.failure.retry";
114
115 private static final String INPUT_FOLDER_PREFIX = "export-files.";
116
117
118 public enum Counter {
119 MISSING_FILES, FILES_COPIED, FILES_SKIPPED, COPY_FAILED,
120 BYTES_EXPECTED, BYTES_SKIPPED, BYTES_COPIED
121 }
122
123 private static class ExportMapper extends Mapper<BytesWritable, NullWritable,
124 NullWritable, NullWritable> {
125 final static int REPORT_SIZE = 1 * 1024 * 1024;
126 final static int BUFFER_SIZE = 64 * 1024;
127
128 private boolean testFailures;
129 private Random random;
130
131 private boolean verifyChecksum;
132 private String filesGroup;
133 private String filesUser;
134 private short filesMode;
135 private int bufferSize;
136
137 private FileSystem outputFs;
138 private Path outputArchive;
139 private Path outputRoot;
140
141 private FileSystem inputFs;
142 private Path inputArchive;
143 private Path inputRoot;
144
145 @Override
146 public void setup(Context context) throws IOException {
147 Configuration conf = context.getConfiguration();
148 Configuration srcConf = HBaseConfiguration.createClusterConf(conf, null, CONF_SOURCE_PREFIX);
149 Configuration destConf = HBaseConfiguration.createClusterConf(conf, null, CONF_DEST_PREFIX);
150
151 verifyChecksum = conf.getBoolean(CONF_CHECKSUM_VERIFY, true);
152
153 filesGroup = conf.get(CONF_FILES_GROUP);
154 filesUser = conf.get(CONF_FILES_USER);
155 filesMode = (short)conf.getInt(CONF_FILES_MODE, 0);
156 outputRoot = new Path(conf.get(CONF_OUTPUT_ROOT));
157 inputRoot = new Path(conf.get(CONF_INPUT_ROOT));
158
159 inputArchive = new Path(inputRoot, HConstants.HFILE_ARCHIVE_DIRECTORY);
160 outputArchive = new Path(outputRoot, HConstants.HFILE_ARCHIVE_DIRECTORY);
161
162 testFailures = conf.getBoolean(CONF_TEST_FAILURE, false);
163
164 try {
165 srcConf.setBoolean("fs." + inputRoot.toUri().getScheme() + ".impl.disable.cache", true);
166 inputFs = FileSystem.get(inputRoot.toUri(), srcConf);
167 } catch (IOException e) {
168 throw new IOException("Could not get the input FileSystem with root=" + inputRoot, e);
169 }
170
171 try {
172 destConf.setBoolean("fs." + outputRoot.toUri().getScheme() + ".impl.disable.cache", true);
173 outputFs = FileSystem.get(outputRoot.toUri(), destConf);
174 } catch (IOException e) {
175 throw new IOException("Could not get the output FileSystem with root="+ outputRoot, e);
176 }
177
178
179 int defaultBlockSize = Math.max((int) outputFs.getDefaultBlockSize(outputRoot), BUFFER_SIZE);
180 bufferSize = conf.getInt(CONF_BUFFER_SIZE, defaultBlockSize);
181 LOG.info("Using bufferSize=" + StringUtils.humanReadableInt(bufferSize));
182
183 for (Counter c : Counter.values()) {
184 context.getCounter(c).increment(0);
185 }
186 }
187
188 @Override
189 protected void cleanup(Context context) {
190 IOUtils.closeStream(inputFs);
191 IOUtils.closeStream(outputFs);
192 }
193
194 @Override
195 public void map(BytesWritable key, NullWritable value, Context context)
196 throws InterruptedException, IOException {
197 SnapshotFileInfo inputInfo = SnapshotFileInfo.parseFrom(key.copyBytes());
198 Path outputPath = getOutputPath(inputInfo);
199
200 copyFile(context, inputInfo, outputPath);
201 }
202
203
204
205
206 private Path getOutputPath(final SnapshotFileInfo inputInfo) throws IOException {
207 Path path = null;
208 switch (inputInfo.getType()) {
209 case HFILE:
210 Path inputPath = new Path(inputInfo.getHfile());
211 String family = inputPath.getParent().getName();
212 TableName table =HFileLink.getReferencedTableName(inputPath.getName());
213 String region = HFileLink.getReferencedRegionName(inputPath.getName());
214 String hfile = HFileLink.getReferencedHFileName(inputPath.getName());
215 path = new Path(FSUtils.getTableDir(new Path("./"), table),
216 new Path(region, new Path(family, hfile)));
217 break;
218 case WAL:
219 LOG.warn("snapshot does not keeps WALs: " + inputInfo);
220 break;
221 default:
222 throw new IOException("Invalid File Type: " + inputInfo.getType().toString());
223 }
224 return new Path(outputArchive, path);
225 }
226
227
228
229
230 private void injectTestFailure(final Context context, final SnapshotFileInfo inputInfo)
231 throws IOException {
232 if (testFailures) {
233 if (context.getConfiguration().getBoolean(CONF_TEST_RETRY, false)) {
234 if (random == null) {
235 random = new Random();
236 }
237
238
239
240
241 if (random.nextFloat() < 0.03) {
242 throw new IOException("TEST RETRY FAILURE: Unable to copy input=" + inputInfo
243 + " time=" + System.currentTimeMillis());
244 }
245 } else {
246 context.getCounter(Counter.COPY_FAILED).increment(1);
247 throw new IOException("TEST FAILURE: Unable to copy input=" + inputInfo);
248 }
249 }
250 }
251
252 private void copyFile(final Context context, final SnapshotFileInfo inputInfo,
253 final Path outputPath) throws IOException {
254 injectTestFailure(context, inputInfo);
255
256
257 FileStatus inputStat = getSourceFileStatus(context, inputInfo);
258
259
260 if (outputFs.exists(outputPath)) {
261 FileStatus outputStat = outputFs.getFileStatus(outputPath);
262 if (outputStat != null && sameFile(inputStat, outputStat)) {
263 LOG.info("Skip copy " + inputStat.getPath() + " to " + outputPath + ", same file.");
264 context.getCounter(Counter.FILES_SKIPPED).increment(1);
265 context.getCounter(Counter.BYTES_SKIPPED).increment(inputStat.getLen());
266 return;
267 }
268 }
269
270 InputStream in = openSourceFile(context, inputInfo);
271 int bandwidthMB = context.getConfiguration().getInt(CONF_BANDWIDTH_MB, 100);
272 if (Integer.MAX_VALUE != bandwidthMB) {
273 in = new ThrottledInputStream(new BufferedInputStream(in), bandwidthMB * 1024L * 1024L);
274 }
275
276 try {
277 context.getCounter(Counter.BYTES_EXPECTED).increment(inputStat.getLen());
278
279
280 createOutputPath(outputPath.getParent());
281 FSDataOutputStream out = outputFs.create(outputPath, true);
282 try {
283 copyData(context, inputStat.getPath(), in, outputPath, out, inputStat.getLen());
284 } finally {
285 out.close();
286 }
287
288
289 if (!preserveAttributes(outputPath, inputStat)) {
290 LOG.warn("You may have to run manually chown on: " + outputPath);
291 }
292 } finally {
293 in.close();
294 }
295 }
296
297
298
299
300 private void createOutputPath(final Path path) throws IOException {
301 if (filesUser == null && filesGroup == null) {
302 outputFs.mkdirs(path);
303 } else {
304 Path parent = path.getParent();
305 if (!outputFs.exists(parent) && !parent.isRoot()) {
306 createOutputPath(parent);
307 }
308 outputFs.mkdirs(path);
309 if (filesUser != null || filesGroup != null) {
310
311 outputFs.setOwner(path, filesUser, filesGroup);
312 }
313 if (filesMode > 0) {
314 outputFs.setPermission(path, new FsPermission(filesMode));
315 }
316 }
317 }
318
319
320
321
322
323
324
325
326
327 private boolean preserveAttributes(final Path path, final FileStatus refStat) {
328 FileStatus stat;
329 try {
330 stat = outputFs.getFileStatus(path);
331 } catch (IOException e) {
332 LOG.warn("Unable to get the status for file=" + path);
333 return false;
334 }
335
336 try {
337 if (filesMode > 0 && stat.getPermission().toShort() != filesMode) {
338 outputFs.setPermission(path, new FsPermission(filesMode));
339 } else if (refStat != null && !stat.getPermission().equals(refStat.getPermission())) {
340 outputFs.setPermission(path, refStat.getPermission());
341 }
342 } catch (IOException e) {
343 LOG.warn("Unable to set the permission for file="+ stat.getPath() +": "+ e.getMessage());
344 return false;
345 }
346
347 boolean hasRefStat = (refStat != null);
348 String user = stringIsNotEmpty(filesUser) || !hasRefStat ? filesUser : refStat.getOwner();
349 String group = stringIsNotEmpty(filesGroup) || !hasRefStat ? filesGroup : refStat.getGroup();
350 if (stringIsNotEmpty(user) || stringIsNotEmpty(group)) {
351 try {
352 if (!(user.equals(stat.getOwner()) && group.equals(stat.getGroup()))) {
353 outputFs.setOwner(path, user, group);
354 }
355 } catch (IOException e) {
356 LOG.warn("Unable to set the owner/group for file="+ stat.getPath() +": "+ e.getMessage());
357 LOG.warn("The user/group may not exist on the destination cluster: user=" +
358 user + " group=" + group);
359 return false;
360 }
361 }
362
363 return true;
364 }
365
366 private boolean stringIsNotEmpty(final String str) {
367 return str != null && str.length() > 0;
368 }
369
370 private void copyData(final Context context,
371 final Path inputPath, final InputStream in,
372 final Path outputPath, final FSDataOutputStream out,
373 final long inputFileSize)
374 throws IOException {
375 final String statusMessage = "copied %s/" + StringUtils.humanReadableInt(inputFileSize) +
376 " (%.1f%%)";
377
378 try {
379 byte[] buffer = new byte[bufferSize];
380 long totalBytesWritten = 0;
381 int reportBytes = 0;
382 int bytesRead;
383
384 long stime = System.currentTimeMillis();
385 while ((bytesRead = in.read(buffer)) > 0) {
386 out.write(buffer, 0, bytesRead);
387 totalBytesWritten += bytesRead;
388 reportBytes += bytesRead;
389
390 if (reportBytes >= REPORT_SIZE) {
391 context.getCounter(Counter.BYTES_COPIED).increment(reportBytes);
392 context.setStatus(String.format(statusMessage,
393 StringUtils.humanReadableInt(totalBytesWritten),
394 (totalBytesWritten/(float)inputFileSize) * 100.0f) +
395 " from " + inputPath + " to " + outputPath);
396 reportBytes = 0;
397 }
398 }
399 long etime = System.currentTimeMillis();
400
401 context.getCounter(Counter.BYTES_COPIED).increment(reportBytes);
402 context.setStatus(String.format(statusMessage,
403 StringUtils.humanReadableInt(totalBytesWritten),
404 (totalBytesWritten/(float)inputFileSize) * 100.0f) +
405 " from " + inputPath + " to " + outputPath);
406
407
408 if (totalBytesWritten != inputFileSize) {
409 String msg = "number of bytes copied not matching copied=" + totalBytesWritten +
410 " expected=" + inputFileSize + " for file=" + inputPath;
411 throw new IOException(msg);
412 }
413
414 LOG.info("copy completed for input=" + inputPath + " output=" + outputPath);
415 LOG.info("size=" + totalBytesWritten +
416 " (" + StringUtils.humanReadableInt(totalBytesWritten) + ")" +
417 " time=" + StringUtils.formatTimeDiff(etime, stime) +
418 String.format(" %.3fM/sec", (totalBytesWritten / ((etime - stime)/1000.0))/1048576.0));
419 context.getCounter(Counter.FILES_COPIED).increment(1);
420 } catch (IOException e) {
421 LOG.error("Error copying " + inputPath + " to " + outputPath, e);
422 context.getCounter(Counter.COPY_FAILED).increment(1);
423 throw e;
424 }
425 }
426
427
428
429
430
431
432 private FSDataInputStream openSourceFile(Context context, final SnapshotFileInfo fileInfo)
433 throws IOException {
434 try {
435 Configuration conf = context.getConfiguration();
436 FileLink link = null;
437 switch (fileInfo.getType()) {
438 case HFILE:
439 Path inputPath = new Path(fileInfo.getHfile());
440 link = HFileLink.buildFromHFileLinkPattern(conf, inputPath);
441 break;
442 case WAL:
443 String serverName = fileInfo.getWalServer();
444 String logName = fileInfo.getWalName();
445 link = new WALLink(inputRoot, serverName, logName);
446 break;
447 default:
448 throw new IOException("Invalid File Type: " + fileInfo.getType().toString());
449 }
450 return link.open(inputFs);
451 } catch (IOException e) {
452 context.getCounter(Counter.MISSING_FILES).increment(1);
453 LOG.error("Unable to open source file=" + fileInfo.toString(), e);
454 throw e;
455 }
456 }
457
458 private FileStatus getSourceFileStatus(Context context, final SnapshotFileInfo fileInfo)
459 throws IOException {
460 try {
461 Configuration conf = context.getConfiguration();
462 FileLink link = null;
463 switch (fileInfo.getType()) {
464 case HFILE:
465 Path inputPath = new Path(fileInfo.getHfile());
466 link = HFileLink.buildFromHFileLinkPattern(conf, inputPath);
467 break;
468 case WAL:
469 link = new WALLink(inputRoot, fileInfo.getWalServer(), fileInfo.getWalName());
470 break;
471 default:
472 throw new IOException("Invalid File Type: " + fileInfo.getType().toString());
473 }
474 return link.getFileStatus(inputFs);
475 } catch (FileNotFoundException e) {
476 context.getCounter(Counter.MISSING_FILES).increment(1);
477 LOG.error("Unable to get the status for source file=" + fileInfo.toString(), e);
478 throw e;
479 } catch (IOException e) {
480 LOG.error("Unable to get the status for source file=" + fileInfo.toString(), e);
481 throw e;
482 }
483 }
484
485 private FileChecksum getFileChecksum(final FileSystem fs, final Path path) {
486 try {
487 return fs.getFileChecksum(path);
488 } catch (IOException e) {
489 LOG.warn("Unable to get checksum for file=" + path, e);
490 return null;
491 }
492 }
493
494
495
496
497
498 private boolean sameFile(final FileStatus inputStat, final FileStatus outputStat) {
499
500 if (inputStat.getLen() != outputStat.getLen()) return false;
501
502
503 if (!verifyChecksum) return true;
504
505
506 FileChecksum inChecksum = getFileChecksum(inputFs, inputStat.getPath());
507 if (inChecksum == null) return false;
508
509 FileChecksum outChecksum = getFileChecksum(outputFs, outputStat.getPath());
510 if (outChecksum == null) return false;
511
512 return inChecksum.equals(outChecksum);
513 }
514 }
515
516
517
518
519
520
521
522
523
524 private static List<Pair<SnapshotFileInfo, Long>> getSnapshotFiles(final Configuration conf,
525 final FileSystem fs, final Path snapshotDir) throws IOException {
526 SnapshotDescription snapshotDesc = SnapshotDescriptionUtils.readSnapshotInfo(fs, snapshotDir);
527
528 final List<Pair<SnapshotFileInfo, Long>> files = new ArrayList<Pair<SnapshotFileInfo, Long>>();
529 final TableName table = TableName.valueOf(snapshotDesc.getTable());
530
531
532 LOG.info("Loading Snapshot '" + snapshotDesc.getName() + "' hfile list");
533 SnapshotReferenceUtil.visitReferencedFiles(conf, fs, snapshotDir, snapshotDesc,
534 new SnapshotReferenceUtil.SnapshotVisitor() {
535 @Override
536 public void storeFile(final HRegionInfo regionInfo, final String family,
537 final SnapshotRegionManifest.StoreFile storeFile) throws IOException {
538 if (storeFile.hasReference()) {
539
540 } else {
541 String region = regionInfo.getEncodedName();
542 String hfile = storeFile.getName();
543 Path path = HFileLink.createPath(table, region, family, hfile);
544
545 SnapshotFileInfo fileInfo = SnapshotFileInfo.newBuilder()
546 .setType(SnapshotFileInfo.Type.HFILE)
547 .setHfile(path.toString())
548 .build();
549
550 long size;
551 if (storeFile.hasFileSize()) {
552 size = storeFile.getFileSize();
553 } else {
554 size = HFileLink.buildFromHFileLinkPattern(conf, path).getFileStatus(fs).getLen();
555 }
556 files.add(new Pair<SnapshotFileInfo, Long>(fileInfo, size));
557 }
558 }
559 });
560
561 return files;
562 }
563
564
565
566
567
568
569
570
571
572 static List<List<Pair<SnapshotFileInfo, Long>>> getBalancedSplits(
573 final List<Pair<SnapshotFileInfo, Long>> files, final int ngroups) {
574
575 Collections.sort(files, new Comparator<Pair<SnapshotFileInfo, Long>>() {
576 public int compare(Pair<SnapshotFileInfo, Long> a, Pair<SnapshotFileInfo, Long> b) {
577 long r = a.getSecond() - b.getSecond();
578 return (r < 0) ? -1 : ((r > 0) ? 1 : 0);
579 }
580 });
581
582
583 List<List<Pair<SnapshotFileInfo, Long>>> fileGroups =
584 new LinkedList<List<Pair<SnapshotFileInfo, Long>>>();
585 long[] sizeGroups = new long[ngroups];
586 int hi = files.size() - 1;
587 int lo = 0;
588
589 List<Pair<SnapshotFileInfo, Long>> group;
590 int dir = 1;
591 int g = 0;
592
593 while (hi >= lo) {
594 if (g == fileGroups.size()) {
595 group = new LinkedList<Pair<SnapshotFileInfo, Long>>();
596 fileGroups.add(group);
597 } else {
598 group = fileGroups.get(g);
599 }
600
601 Pair<SnapshotFileInfo, Long> fileInfo = files.get(hi--);
602
603
604 sizeGroups[g] += fileInfo.getSecond();
605 group.add(fileInfo);
606
607
608 g += dir;
609 if (g == ngroups) {
610 dir = -1;
611 g = ngroups - 1;
612 } else if (g < 0) {
613 dir = 1;
614 g = 0;
615 }
616 }
617
618 if (LOG.isDebugEnabled()) {
619 for (int i = 0; i < sizeGroups.length; ++i) {
620 LOG.debug("export split=" + i + " size=" + StringUtils.humanReadableInt(sizeGroups[i]));
621 }
622 }
623
624 return fileGroups;
625 }
626
627 private static class ExportSnapshotInputFormat extends InputFormat<BytesWritable, NullWritable> {
628 @Override
629 public RecordReader<BytesWritable, NullWritable> createRecordReader(InputSplit split,
630 TaskAttemptContext tac) throws IOException, InterruptedException {
631 return new ExportSnapshotRecordReader(((ExportSnapshotInputSplit)split).getSplitKeys());
632 }
633
634 @Override
635 public List<InputSplit> getSplits(JobContext context) throws IOException, InterruptedException {
636 Configuration conf = context.getConfiguration();
637 Path snapshotDir = new Path(conf.get(CONF_SNAPSHOT_DIR));
638 FileSystem fs = FileSystem.get(snapshotDir.toUri(), conf);
639
640 List<Pair<SnapshotFileInfo, Long>> snapshotFiles = getSnapshotFiles(conf, fs, snapshotDir);
641 int mappers = conf.getInt(CONF_NUM_SPLITS, 0);
642 if (mappers == 0 && snapshotFiles.size() > 0) {
643 mappers = 1 + (snapshotFiles.size() / conf.getInt(CONF_MAP_GROUP, 10));
644 mappers = Math.min(mappers, snapshotFiles.size());
645 conf.setInt(CONF_NUM_SPLITS, mappers);
646 conf.setInt(MR_NUM_MAPS, mappers);
647 }
648
649 List<List<Pair<SnapshotFileInfo, Long>>> groups = getBalancedSplits(snapshotFiles, mappers);
650 List<InputSplit> splits = new ArrayList(groups.size());
651 for (List<Pair<SnapshotFileInfo, Long>> files: groups) {
652 splits.add(new ExportSnapshotInputSplit(files));
653 }
654 return splits;
655 }
656
657 private static class ExportSnapshotInputSplit extends InputSplit implements Writable {
658 private List<Pair<BytesWritable, Long>> files;
659 private long length;
660
661 public ExportSnapshotInputSplit() {
662 this.files = null;
663 }
664
665 public ExportSnapshotInputSplit(final List<Pair<SnapshotFileInfo, Long>> snapshotFiles) {
666 this.files = new ArrayList(snapshotFiles.size());
667 for (Pair<SnapshotFileInfo, Long> fileInfo: snapshotFiles) {
668 this.files.add(new Pair<BytesWritable, Long>(
669 new BytesWritable(fileInfo.getFirst().toByteArray()), fileInfo.getSecond()));
670 this.length += fileInfo.getSecond();
671 }
672 }
673
674 private List<Pair<BytesWritable, Long>> getSplitKeys() {
675 return files;
676 }
677
678 @Override
679 public long getLength() throws IOException, InterruptedException {
680 return length;
681 }
682
683 @Override
684 public String[] getLocations() throws IOException, InterruptedException {
685 return new String[] {};
686 }
687
688 @Override
689 public void readFields(DataInput in) throws IOException {
690 int count = in.readInt();
691 files = new ArrayList<Pair<BytesWritable, Long>>(count);
692 length = 0;
693 for (int i = 0; i < count; ++i) {
694 BytesWritable fileInfo = new BytesWritable();
695 fileInfo.readFields(in);
696 long size = in.readLong();
697 files.add(new Pair<BytesWritable, Long>(fileInfo, size));
698 length += size;
699 }
700 }
701
702 @Override
703 public void write(DataOutput out) throws IOException {
704 out.writeInt(files.size());
705 for (final Pair<BytesWritable, Long> fileInfo: files) {
706 fileInfo.getFirst().write(out);
707 out.writeLong(fileInfo.getSecond());
708 }
709 }
710 }
711
712 private static class ExportSnapshotRecordReader
713 extends RecordReader<BytesWritable, NullWritable> {
714 private final List<Pair<BytesWritable, Long>> files;
715 private long totalSize = 0;
716 private long procSize = 0;
717 private int index = -1;
718
719 ExportSnapshotRecordReader(final List<Pair<BytesWritable, Long>> files) {
720 this.files = files;
721 for (Pair<BytesWritable, Long> fileInfo: files) {
722 totalSize += fileInfo.getSecond();
723 }
724 }
725
726 @Override
727 public void close() { }
728
729 @Override
730 public BytesWritable getCurrentKey() { return files.get(index).getFirst(); }
731
732 @Override
733 public NullWritable getCurrentValue() { return NullWritable.get(); }
734
735 @Override
736 public float getProgress() { return (float)procSize / totalSize; }
737
738 @Override
739 public void initialize(InputSplit split, TaskAttemptContext tac) { }
740
741 @Override
742 public boolean nextKeyValue() {
743 if (index >= 0) {
744 procSize += files.get(index).getSecond();
745 }
746 return(++index < files.size());
747 }
748 }
749 }
750
751
752
753
754
755
756
757
758 private void runCopyJob(final Path inputRoot, final Path outputRoot,
759 final String snapshotName, final Path snapshotDir, final boolean verifyChecksum,
760 final String filesUser, final String filesGroup, final int filesMode,
761 final int mappers, final int bandwidthMB)
762 throws IOException, InterruptedException, ClassNotFoundException {
763 Configuration conf = getConf();
764 if (filesGroup != null) conf.set(CONF_FILES_GROUP, filesGroup);
765 if (filesUser != null) conf.set(CONF_FILES_USER, filesUser);
766 if (mappers > 0) {
767 conf.setInt(CONF_NUM_SPLITS, mappers);
768 conf.setInt(MR_NUM_MAPS, mappers);
769 }
770 conf.setInt(CONF_FILES_MODE, filesMode);
771 conf.setBoolean(CONF_CHECKSUM_VERIFY, verifyChecksum);
772 conf.set(CONF_OUTPUT_ROOT, outputRoot.toString());
773 conf.set(CONF_INPUT_ROOT, inputRoot.toString());
774 conf.setInt(CONF_BANDWIDTH_MB, bandwidthMB);
775 conf.set(CONF_SNAPSHOT_NAME, snapshotName);
776 conf.set(CONF_SNAPSHOT_DIR, snapshotDir.toString());
777
778 Job job = new Job(conf);
779 job.setJobName("ExportSnapshot-" + snapshotName);
780 job.setJarByClass(ExportSnapshot.class);
781 TableMapReduceUtil.addDependencyJars(job);
782 job.setMapperClass(ExportMapper.class);
783 job.setInputFormatClass(ExportSnapshotInputFormat.class);
784 job.setOutputFormatClass(NullOutputFormat.class);
785 job.setMapSpeculativeExecution(false);
786 job.setNumReduceTasks(0);
787
788
789 Configuration srcConf = HBaseConfiguration.createClusterConf(conf, null, CONF_SOURCE_PREFIX);
790 TokenCache.obtainTokensForNamenodes(job.getCredentials(),
791 new Path[] { inputRoot }, srcConf);
792 Configuration destConf = HBaseConfiguration.createClusterConf(conf, null, CONF_DEST_PREFIX);
793 TokenCache.obtainTokensForNamenodes(job.getCredentials(),
794 new Path[] { outputRoot }, destConf);
795
796
797 if (!job.waitForCompletion(true)) {
798 throw new ExportSnapshotException(job.getStatus().getFailureInfo());
799 }
800 }
801
802 private void verifySnapshot(final Configuration baseConf,
803 final FileSystem fs, final Path rootDir, final Path snapshotDir) throws IOException {
804
805 Configuration conf = new Configuration(baseConf);
806 FSUtils.setRootDir(conf, rootDir);
807 FSUtils.setFsDefault(conf, FSUtils.getRootDir(conf));
808 SnapshotDescription snapshotDesc = SnapshotDescriptionUtils.readSnapshotInfo(fs, snapshotDir);
809 SnapshotReferenceUtil.verifySnapshot(conf, fs, snapshotDir, snapshotDesc);
810 }
811
812
813
814
815 private void setOwner(final FileSystem fs, final Path path, final String user,
816 final String group, final boolean recursive) throws IOException {
817 if (user != null || group != null) {
818 if (recursive && fs.isDirectory(path)) {
819 for (FileStatus child : fs.listStatus(path)) {
820 setOwner(fs, child.getPath(), user, group, recursive);
821 }
822 }
823 fs.setOwner(path, user, group);
824 }
825 }
826
827
828
829
830 private void setPermission(final FileSystem fs, final Path path, final short filesMode,
831 final boolean recursive) throws IOException {
832 if (filesMode > 0) {
833 FsPermission perm = new FsPermission(filesMode);
834 if (recursive && fs.isDirectory(path)) {
835 for (FileStatus child : fs.listStatus(path)) {
836 setPermission(fs, child.getPath(), filesMode, recursive);
837 }
838 }
839 fs.setPermission(path, perm);
840 }
841 }
842
843
844
845
846
847 @Override
848 public int run(String[] args) throws IOException {
849 boolean verifyTarget = true;
850 boolean verifyChecksum = true;
851 String snapshotName = null;
852 String targetName = null;
853 boolean overwrite = false;
854 String filesGroup = null;
855 String filesUser = null;
856 Path outputRoot = null;
857 int bandwidthMB = Integer.MAX_VALUE;
858 int filesMode = 0;
859 int mappers = 0;
860
861 Configuration conf = getConf();
862 Path inputRoot = FSUtils.getRootDir(conf);
863
864
865 for (int i = 0; i < args.length; i++) {
866 String cmd = args[i];
867 if (cmd.equals("-snapshot")) {
868 snapshotName = args[++i];
869 } else if (cmd.equals("-target")) {
870 targetName = args[++i];
871 } else if (cmd.equals("-copy-to")) {
872 outputRoot = new Path(args[++i]);
873 } else if (cmd.equals("-copy-from")) {
874 inputRoot = new Path(args[++i]);
875 FSUtils.setRootDir(conf, inputRoot);
876 } else if (cmd.equals("-no-checksum-verify")) {
877 verifyChecksum = false;
878 } else if (cmd.equals("-no-target-verify")) {
879 verifyTarget = false;
880 } else if (cmd.equals("-mappers")) {
881 mappers = Integer.parseInt(args[++i]);
882 } else if (cmd.equals("-chuser")) {
883 filesUser = args[++i];
884 } else if (cmd.equals("-chgroup")) {
885 filesGroup = args[++i];
886 } else if (cmd.equals("-bandwidth")) {
887 bandwidthMB = Integer.parseInt(args[++i]);
888 } else if (cmd.equals("-chmod")) {
889 filesMode = Integer.parseInt(args[++i], 8);
890 } else if (cmd.equals("-overwrite")) {
891 overwrite = true;
892 } else if (cmd.equals("-h") || cmd.equals("--help")) {
893 printUsageAndExit();
894 } else {
895 System.err.println("UNEXPECTED: " + cmd);
896 printUsageAndExit();
897 }
898 }
899
900
901 if (snapshotName == null) {
902 System.err.println("Snapshot name not provided.");
903 printUsageAndExit();
904 }
905
906 if (outputRoot == null) {
907 System.err.println("Destination file-system not provided.");
908 printUsageAndExit();
909 }
910
911 if (targetName == null) {
912 targetName = snapshotName;
913 }
914
915 Configuration srcConf = HBaseConfiguration.createClusterConf(conf, null, CONF_SOURCE_PREFIX);
916 srcConf.setBoolean("fs." + inputRoot.toUri().getScheme() + ".impl.disable.cache", true);
917 FileSystem inputFs = FileSystem.get(inputRoot.toUri(), srcConf);
918 LOG.debug("inputFs=" + inputFs.getUri().toString() + " inputRoot=" + inputRoot);
919 Configuration destConf = HBaseConfiguration.createClusterConf(conf, null, CONF_DEST_PREFIX);
920 destConf.setBoolean("fs." + outputRoot.toUri().getScheme() + ".impl.disable.cache", true);
921 FileSystem outputFs = FileSystem.get(outputRoot.toUri(), destConf);
922 LOG.debug("outputFs=" + outputFs.getUri().toString() + " outputRoot=" + outputRoot.toString());
923
924 boolean skipTmp = conf.getBoolean(CONF_SKIP_TMP, false);
925
926 Path snapshotDir = SnapshotDescriptionUtils.getCompletedSnapshotDir(snapshotName, inputRoot);
927 Path snapshotTmpDir = SnapshotDescriptionUtils.getWorkingSnapshotDir(targetName, outputRoot);
928 Path outputSnapshotDir = SnapshotDescriptionUtils.getCompletedSnapshotDir(targetName, outputRoot);
929 Path initialOutputSnapshotDir = skipTmp ? outputSnapshotDir : snapshotTmpDir;
930
931
932 Path needSetOwnerDir = SnapshotDescriptionUtils.getSnapshotRootDir(outputRoot);
933 if (outputFs.exists(needSetOwnerDir)) {
934 if (skipTmp) {
935 needSetOwnerDir = outputSnapshotDir;
936 } else {
937 needSetOwnerDir = SnapshotDescriptionUtils.getWorkingSnapshotDir(outputRoot);
938 if (outputFs.exists(needSetOwnerDir)) {
939 needSetOwnerDir = snapshotTmpDir;
940 }
941 }
942 }
943
944
945 if (outputFs.exists(outputSnapshotDir)) {
946 if (overwrite) {
947 if (!outputFs.delete(outputSnapshotDir, true)) {
948 System.err.println("Unable to remove existing snapshot directory: " + outputSnapshotDir);
949 return 1;
950 }
951 } else {
952 System.err.println("The snapshot '" + targetName +
953 "' already exists in the destination: " + outputSnapshotDir);
954 return 1;
955 }
956 }
957
958 if (!skipTmp) {
959
960 if (outputFs.exists(snapshotTmpDir)) {
961 if (overwrite) {
962 if (!outputFs.delete(snapshotTmpDir, true)) {
963 System.err.println("Unable to remove existing snapshot tmp directory: "+snapshotTmpDir);
964 return 1;
965 }
966 } else {
967 System.err.println("A snapshot with the same name '"+ targetName +"' may be in-progress");
968 System.err.println("Please check "+snapshotTmpDir+". If the snapshot has completed, ");
969 System.err.println("consider removing "+snapshotTmpDir+" by using the -overwrite option");
970 return 1;
971 }
972 }
973 }
974
975
976
977
978 try {
979 LOG.info("Copy Snapshot Manifest");
980 FileUtil.copy(inputFs, snapshotDir, outputFs, initialOutputSnapshotDir, false, false, conf);
981 } catch (IOException e) {
982 throw new ExportSnapshotException("Failed to copy the snapshot directory: from=" +
983 snapshotDir + " to=" + initialOutputSnapshotDir, e);
984 } finally {
985 if (filesUser != null || filesGroup != null) {
986 LOG.warn((filesUser == null ? "" : "Change the owner of " + needSetOwnerDir + " to "
987 + filesUser)
988 + (filesGroup == null ? "" : ", Change the group of " + needSetOwnerDir + " to "
989 + filesGroup));
990 setOwner(outputFs, needSetOwnerDir, filesUser, filesGroup, true);
991 }
992 if (filesMode > 0) {
993 LOG.warn("Change the permission of " + needSetOwnerDir + " to " + filesMode);
994 setPermission(outputFs, needSetOwnerDir, (short)filesMode, true);
995 }
996 }
997
998
999 if (!targetName.equals(snapshotName)) {
1000 SnapshotDescription snapshotDesc =
1001 SnapshotDescriptionUtils.readSnapshotInfo(inputFs, snapshotDir)
1002 .toBuilder()
1003 .setName(targetName)
1004 .build();
1005 SnapshotDescriptionUtils.writeSnapshotInfo(snapshotDesc, snapshotTmpDir, outputFs);
1006 }
1007
1008
1009
1010
1011 try {
1012 runCopyJob(inputRoot, outputRoot, snapshotName, snapshotDir, verifyChecksum,
1013 filesUser, filesGroup, filesMode, mappers, bandwidthMB);
1014
1015 LOG.info("Finalize the Snapshot Export");
1016 if (!skipTmp) {
1017
1018 if (!outputFs.rename(snapshotTmpDir, outputSnapshotDir)) {
1019 throw new ExportSnapshotException("Unable to rename snapshot directory from=" +
1020 snapshotTmpDir + " to=" + outputSnapshotDir);
1021 }
1022 }
1023
1024
1025 if (verifyTarget) {
1026 LOG.info("Verify snapshot integrity");
1027 verifySnapshot(destConf, outputFs, outputRoot, outputSnapshotDir);
1028 }
1029
1030 LOG.info("Export Completed: " + targetName);
1031 return 0;
1032 } catch (Exception e) {
1033 LOG.error("Snapshot export failed", e);
1034 if (!skipTmp) {
1035 outputFs.delete(snapshotTmpDir, true);
1036 }
1037 outputFs.delete(outputSnapshotDir, true);
1038 return 1;
1039 } finally {
1040 IOUtils.closeStream(inputFs);
1041 IOUtils.closeStream(outputFs);
1042 }
1043 }
1044
1045
1046 private void printUsageAndExit() {
1047 System.err.printf("Usage: bin/hbase %s [options]%n", getClass().getName());
1048 System.err.println(" where [options] are:");
1049 System.err.println(" -h|-help Show this help and exit.");
1050 System.err.println(" -snapshot NAME Snapshot to restore.");
1051 System.err.println(" -copy-to NAME Remote destination hdfs://");
1052 System.err.println(" -copy-from NAME Input folder hdfs:// (default hbase.rootdir)");
1053 System.err.println(" -no-checksum-verify Do not verify checksum, use name+length only.");
1054 System.err.println(" -no-target-verify Do not verify the integrity of the \\" +
1055 "exported snapshot.");
1056 System.err.println(" -overwrite Rewrite the snapshot manifest if already exists");
1057 System.err.println(" -chuser USERNAME Change the owner of the files " +
1058 "to the specified one.");
1059 System.err.println(" -chgroup GROUP Change the group of the files to " +
1060 "the specified one.");
1061 System.err.println(" -chmod MODE Change the permission of the files " +
1062 "to the specified one.");
1063 System.err.println(" -mappers Number of mappers to use during the " +
1064 "copy (mapreduce.job.maps).");
1065 System.err.println(" -bandwidth Limit bandwidth to this value in MB/second.");
1066 System.err.println();
1067 System.err.println("Examples:");
1068 System.err.println(" hbase snapshot export \\");
1069 System.err.println(" -snapshot MySnapshot -copy-to hdfs://srv2:8082/hbase \\");
1070 System.err.println(" -chuser MyUser -chgroup MyGroup -chmod 700 -mappers 16");
1071 System.err.println();
1072 System.err.println(" hbase snapshot export \\");
1073 System.err.println(" -snapshot MySnapshot -copy-from hdfs://srv2:8082/hbase \\");
1074 System.err.println(" -copy-to hdfs://srv1:50070/hbase \\");
1075 System.exit(1);
1076 }
1077
1078
1079
1080
1081
1082
1083
1084
1085 static int innerMain(final Configuration conf, final String [] args) throws Exception {
1086 return ToolRunner.run(conf, new ExportSnapshot(), args);
1087 }
1088
1089 public static void main(String[] args) throws Exception {
1090 System.exit(innerMain(HBaseConfiguration.create(), args));
1091 }
1092 }