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