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
799
800 throw new ExportSnapshotException("Copy Files Map-Reduce Job failed");
801 }
802 }
803
804 private void verifySnapshot(final Configuration baseConf,
805 final FileSystem fs, final Path rootDir, final Path snapshotDir) throws IOException {
806
807 Configuration conf = new Configuration(baseConf);
808 FSUtils.setRootDir(conf, rootDir);
809 FSUtils.setFsDefault(conf, FSUtils.getRootDir(conf));
810 SnapshotDescription snapshotDesc = SnapshotDescriptionUtils.readSnapshotInfo(fs, snapshotDir);
811 SnapshotReferenceUtil.verifySnapshot(conf, fs, snapshotDir, snapshotDesc);
812 }
813
814
815
816
817 private void setOwner(final FileSystem fs, final Path path, final String user,
818 final String group, final boolean recursive) throws IOException {
819 if (user != null || group != null) {
820 if (recursive && fs.isDirectory(path)) {
821 for (FileStatus child : fs.listStatus(path)) {
822 setOwner(fs, child.getPath(), user, group, recursive);
823 }
824 }
825 fs.setOwner(path, user, group);
826 }
827 }
828
829
830
831
832 private void setPermission(final FileSystem fs, final Path path, final short filesMode,
833 final boolean recursive) throws IOException {
834 if (filesMode > 0) {
835 FsPermission perm = new FsPermission(filesMode);
836 if (recursive && fs.isDirectory(path)) {
837 for (FileStatus child : fs.listStatus(path)) {
838 setPermission(fs, child.getPath(), filesMode, recursive);
839 }
840 }
841 fs.setPermission(path, perm);
842 }
843 }
844
845
846
847
848
849 @Override
850 public int run(String[] args) throws IOException {
851 boolean verifyTarget = true;
852 boolean verifyChecksum = true;
853 String snapshotName = null;
854 String targetName = null;
855 boolean overwrite = false;
856 String filesGroup = null;
857 String filesUser = null;
858 Path outputRoot = null;
859 int bandwidthMB = Integer.MAX_VALUE;
860 int filesMode = 0;
861 int mappers = 0;
862
863 Configuration conf = getConf();
864 Path inputRoot = FSUtils.getRootDir(conf);
865
866
867 for (int i = 0; i < args.length; i++) {
868 String cmd = args[i];
869 if (cmd.equals("-snapshot")) {
870 snapshotName = args[++i];
871 } else if (cmd.equals("-target")) {
872 targetName = args[++i];
873 } else if (cmd.equals("-copy-to")) {
874 outputRoot = new Path(args[++i]);
875 } else if (cmd.equals("-copy-from")) {
876 inputRoot = new Path(args[++i]);
877 FSUtils.setRootDir(conf, inputRoot);
878 } else if (cmd.equals("-no-checksum-verify")) {
879 verifyChecksum = false;
880 } else if (cmd.equals("-no-target-verify")) {
881 verifyTarget = false;
882 } else if (cmd.equals("-mappers")) {
883 mappers = Integer.parseInt(args[++i]);
884 } else if (cmd.equals("-chuser")) {
885 filesUser = args[++i];
886 } else if (cmd.equals("-chgroup")) {
887 filesGroup = args[++i];
888 } else if (cmd.equals("-bandwidth")) {
889 bandwidthMB = Integer.parseInt(args[++i]);
890 } else if (cmd.equals("-chmod")) {
891 filesMode = Integer.parseInt(args[++i], 8);
892 } else if (cmd.equals("-overwrite")) {
893 overwrite = true;
894 } else if (cmd.equals("-h") || cmd.equals("--help")) {
895 printUsageAndExit();
896 } else {
897 System.err.println("UNEXPECTED: " + cmd);
898 printUsageAndExit();
899 }
900 }
901
902
903 if (snapshotName == null) {
904 System.err.println("Snapshot name not provided.");
905 printUsageAndExit();
906 }
907
908 if (outputRoot == null) {
909 System.err.println("Destination file-system not provided.");
910 printUsageAndExit();
911 }
912
913 if (targetName == null) {
914 targetName = snapshotName;
915 }
916
917 Configuration srcConf = HBaseConfiguration.createClusterConf(conf, null, CONF_SOURCE_PREFIX);
918 srcConf.setBoolean("fs." + inputRoot.toUri().getScheme() + ".impl.disable.cache", true);
919 FileSystem inputFs = FileSystem.get(inputRoot.toUri(), srcConf);
920 LOG.debug("inputFs=" + inputFs.getUri().toString() + " inputRoot=" + inputRoot);
921 Configuration destConf = HBaseConfiguration.createClusterConf(conf, null, CONF_DEST_PREFIX);
922 destConf.setBoolean("fs." + outputRoot.toUri().getScheme() + ".impl.disable.cache", true);
923 FileSystem outputFs = FileSystem.get(outputRoot.toUri(), destConf);
924 LOG.debug("outputFs=" + outputFs.getUri().toString() + " outputRoot=" + outputRoot.toString());
925
926 boolean skipTmp = conf.getBoolean(CONF_SKIP_TMP, false);
927
928 Path snapshotDir = SnapshotDescriptionUtils.getCompletedSnapshotDir(snapshotName, inputRoot);
929 Path snapshotTmpDir = SnapshotDescriptionUtils.getWorkingSnapshotDir(targetName, outputRoot);
930 Path outputSnapshotDir = SnapshotDescriptionUtils.getCompletedSnapshotDir(targetName, outputRoot);
931 Path initialOutputSnapshotDir = skipTmp ? outputSnapshotDir : snapshotTmpDir;
932
933
934 if (outputFs.exists(outputSnapshotDir)) {
935 if (overwrite) {
936 if (!outputFs.delete(outputSnapshotDir, true)) {
937 System.err.println("Unable to remove existing snapshot directory: " + outputSnapshotDir);
938 return 1;
939 }
940 } else {
941 System.err.println("The snapshot '" + targetName +
942 "' already exists in the destination: " + outputSnapshotDir);
943 return 1;
944 }
945 }
946
947 if (!skipTmp) {
948
949 if (outputFs.exists(snapshotTmpDir)) {
950 if (overwrite) {
951 if (!outputFs.delete(snapshotTmpDir, true)) {
952 System.err.println("Unable to remove existing snapshot tmp directory: "+snapshotTmpDir);
953 return 1;
954 }
955 } else {
956 System.err.println("A snapshot with the same name '"+ targetName +"' may be in-progress");
957 System.err.println("Please check "+snapshotTmpDir+". If the snapshot has completed, ");
958 System.err.println("consider removing "+snapshotTmpDir+" by using the -overwrite option");
959 return 1;
960 }
961 }
962 }
963
964
965
966
967 try {
968 LOG.info("Copy Snapshot Manifest");
969 FileUtil.copy(inputFs, snapshotDir, outputFs, initialOutputSnapshotDir, false, false, conf);
970 if (filesUser != null || filesGroup != null) {
971 setOwner(outputFs, snapshotTmpDir, filesUser, filesGroup, true);
972 }
973 if (filesMode > 0) {
974 setPermission(outputFs, snapshotTmpDir, (short)filesMode, true);
975 }
976 } catch (IOException e) {
977 throw new ExportSnapshotException("Failed to copy the snapshot directory: from=" +
978 snapshotDir + " to=" + initialOutputSnapshotDir, e);
979 }
980
981
982 if (!targetName.equals(snapshotName)) {
983 SnapshotDescription snapshotDesc =
984 SnapshotDescriptionUtils.readSnapshotInfo(inputFs, snapshotDir)
985 .toBuilder()
986 .setName(targetName)
987 .build();
988 SnapshotDescriptionUtils.writeSnapshotInfo(snapshotDesc, snapshotTmpDir, outputFs);
989 }
990
991
992
993
994 try {
995 runCopyJob(inputRoot, outputRoot, snapshotName, snapshotDir, verifyChecksum,
996 filesUser, filesGroup, filesMode, mappers, bandwidthMB);
997
998 LOG.info("Finalize the Snapshot Export");
999 if (!skipTmp) {
1000
1001 if (!outputFs.rename(snapshotTmpDir, outputSnapshotDir)) {
1002 throw new ExportSnapshotException("Unable to rename snapshot directory from=" +
1003 snapshotTmpDir + " to=" + outputSnapshotDir);
1004 }
1005 }
1006
1007
1008 if (verifyTarget) {
1009 LOG.info("Verify snapshot integrity");
1010 verifySnapshot(destConf, outputFs, outputRoot, outputSnapshotDir);
1011 }
1012
1013 LOG.info("Export Completed: " + targetName);
1014 return 0;
1015 } catch (Exception e) {
1016 LOG.error("Snapshot export failed", e);
1017 if (!skipTmp) {
1018 outputFs.delete(snapshotTmpDir, true);
1019 }
1020 outputFs.delete(outputSnapshotDir, true);
1021 return 1;
1022 } finally {
1023 IOUtils.closeStream(inputFs);
1024 IOUtils.closeStream(outputFs);
1025 }
1026 }
1027
1028
1029 private void printUsageAndExit() {
1030 System.err.printf("Usage: bin/hbase %s [options]%n", getClass().getName());
1031 System.err.println(" where [options] are:");
1032 System.err.println(" -h|-help Show this help and exit.");
1033 System.err.println(" -snapshot NAME Snapshot to restore.");
1034 System.err.println(" -copy-to NAME Remote destination hdfs://");
1035 System.err.println(" -copy-from NAME Input folder hdfs:// (default hbase.rootdir)");
1036 System.err.println(" -no-checksum-verify Do not verify checksum, use name+length only.");
1037 System.err.println(" -no-target-verify Do not verify the integrity of the \\" +
1038 "exported snapshot.");
1039 System.err.println(" -overwrite Rewrite the snapshot manifest if already exists");
1040 System.err.println(" -chuser USERNAME Change the owner of the files " +
1041 "to the specified one.");
1042 System.err.println(" -chgroup GROUP Change the group of the files to " +
1043 "the specified one.");
1044 System.err.println(" -chmod MODE Change the permission of the files " +
1045 "to the specified one.");
1046 System.err.println(" -mappers Number of mappers to use during the " +
1047 "copy (mapreduce.job.maps).");
1048 System.err.println(" -bandwidth Limit bandwidth to this value in MB/second.");
1049 System.err.println();
1050 System.err.println("Examples:");
1051 System.err.println(" hbase snapshot export \\");
1052 System.err.println(" -snapshot MySnapshot -copy-to hdfs://srv2:8082/hbase \\");
1053 System.err.println(" -chuser MyUser -chgroup MyGroup -chmod 700 -mappers 16");
1054 System.err.println();
1055 System.err.println(" hbase snapshot export \\");
1056 System.err.println(" -snapshot MySnapshot -copy-from hdfs://srv2:8082/hbase \\");
1057 System.err.println(" -copy-to hdfs://srv1:50070/hbase \\");
1058 System.exit(1);
1059 }
1060
1061
1062
1063
1064
1065
1066
1067
1068 static int innerMain(final Configuration conf, final String [] args) throws Exception {
1069 return ToolRunner.run(conf, new ExportSnapshot(), args);
1070 }
1071
1072 public static void main(String[] args) throws Exception {
1073 System.exit(innerMain(HBaseConfiguration.create(), args));
1074 }
1075 }