1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18 package org.apache.hadoop.hbase.mapreduce;
19
20 import java.io.IOException;
21 import java.util.Iterator;
22
23 import org.apache.commons.logging.Log;
24 import org.apache.commons.logging.LogFactory;
25 import org.apache.hadoop.conf.Configuration;
26 import org.apache.hadoop.conf.Configured;
27 import org.apache.hadoop.fs.FileStatus;
28 import org.apache.hadoop.fs.FileSystem;
29 import org.apache.hadoop.fs.Path;
30 import org.apache.hadoop.hbase.Cell;
31 import org.apache.hadoop.hbase.CellComparator;
32 import org.apache.hadoop.hbase.CellUtil;
33 import org.apache.hadoop.hbase.HBaseConfiguration;
34 import org.apache.hadoop.hbase.TableName;
35 import org.apache.hadoop.hbase.client.Connection;
36 import org.apache.hadoop.hbase.client.ConnectionFactory;
37 import org.apache.hadoop.hbase.client.Delete;
38 import org.apache.hadoop.hbase.client.Mutation;
39 import org.apache.hadoop.hbase.client.Put;
40 import org.apache.hadoop.hbase.client.Result;
41 import org.apache.hadoop.hbase.client.ResultScanner;
42 import org.apache.hadoop.hbase.client.Scan;
43 import org.apache.hadoop.hbase.client.Table;
44 import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
45 import org.apache.hadoop.hbase.util.Bytes;
46 import org.apache.hadoop.mapreduce.Counters;
47 import org.apache.hadoop.mapreduce.Job;
48 import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;
49 import org.apache.hadoop.util.GenericOptionsParser;
50 import org.apache.hadoop.util.Tool;
51 import org.apache.hadoop.util.ToolRunner;
52
53 import com.google.common.base.Throwables;
54 import com.google.common.collect.Iterators;
55
56 public class SyncTable extends Configured implements Tool {
57
58 private static final Log LOG = LogFactory.getLog(SyncTable.class);
59
60 static final String SOURCE_HASH_DIR_CONF_KEY = "sync.table.source.hash.dir";
61 static final String SOURCE_TABLE_CONF_KEY = "sync.table.source.table.name";
62 static final String TARGET_TABLE_CONF_KEY = "sync.table.target.table.name";
63 static final String SOURCE_ZK_CLUSTER_CONF_KEY = "sync.table.source.zk.cluster";
64 static final String TARGET_ZK_CLUSTER_CONF_KEY = "sync.table.target.zk.cluster";
65 static final String DRY_RUN_CONF_KEY="sync.table.dry.run";
66
67 Path sourceHashDir;
68 String sourceTableName;
69 String targetTableName;
70
71 String sourceZkCluster;
72 String targetZkCluster;
73 boolean dryRun;
74
75 Counters counters;
76
77 public SyncTable(Configuration conf) {
78 super(conf);
79 }
80
81 public Job createSubmittableJob(String[] args) throws IOException {
82 FileSystem fs = sourceHashDir.getFileSystem(getConf());
83 if (!fs.exists(sourceHashDir)) {
84 throw new IOException("Source hash dir not found: " + sourceHashDir);
85 }
86
87 HashTable.TableHash tableHash = HashTable.TableHash.read(getConf(), sourceHashDir);
88 LOG.info("Read source hash manifest: " + tableHash);
89 LOG.info("Read " + tableHash.partitions.size() + " partition keys");
90 if (!tableHash.tableName.equals(sourceTableName)) {
91 LOG.warn("Table name mismatch - manifest indicates hash was taken from: "
92 + tableHash.tableName + " but job is reading from: " + sourceTableName);
93 }
94 if (tableHash.numHashFiles != tableHash.partitions.size() + 1) {
95 throw new RuntimeException("Hash data appears corrupt. The number of of hash files created"
96 + " should be 1 more than the number of partition keys. However, the manifest file "
97 + " says numHashFiles=" + tableHash.numHashFiles + " but the number of partition keys"
98 + " found in the partitions file is " + tableHash.partitions.size());
99 }
100
101 Path dataDir = new Path(sourceHashDir, HashTable.HASH_DATA_DIR);
102 int dataSubdirCount = 0;
103 for (FileStatus file : fs.listStatus(dataDir)) {
104 if (file.getPath().getName().startsWith(HashTable.OUTPUT_DATA_FILE_PREFIX)) {
105 dataSubdirCount++;
106 }
107 }
108
109 if (dataSubdirCount != tableHash.numHashFiles) {
110 throw new RuntimeException("Hash data appears corrupt. The number of of hash files created"
111 + " should be 1 more than the number of partition keys. However, the number of data dirs"
112 + " found is " + dataSubdirCount + " but the number of partition keys"
113 + " found in the partitions file is " + tableHash.partitions.size());
114 }
115
116 Job job = Job.getInstance(getConf(),getConf().get("mapreduce.job.name",
117 "syncTable_" + sourceTableName + "-" + targetTableName));
118 Configuration jobConf = job.getConfiguration();
119 job.setJarByClass(HashTable.class);
120 jobConf.set(SOURCE_HASH_DIR_CONF_KEY, sourceHashDir.toString());
121 jobConf.set(SOURCE_TABLE_CONF_KEY, sourceTableName);
122 jobConf.set(TARGET_TABLE_CONF_KEY, targetTableName);
123 if (sourceZkCluster != null) {
124 jobConf.set(SOURCE_ZK_CLUSTER_CONF_KEY, sourceZkCluster);
125 }
126 if (targetZkCluster != null) {
127 jobConf.set(TARGET_ZK_CLUSTER_CONF_KEY, targetZkCluster);
128 }
129 jobConf.setBoolean(DRY_RUN_CONF_KEY, dryRun);
130
131 TableMapReduceUtil.initTableMapperJob(targetTableName, tableHash.initScan(),
132 SyncMapper.class, null, null, job);
133
134 job.setNumReduceTasks(0);
135
136 if (dryRun) {
137 job.setOutputFormatClass(NullOutputFormat.class);
138 } else {
139
140
141 TableMapReduceUtil.initTableReducerJob(targetTableName, null, job, null,
142 targetZkCluster, null, null);
143
144
145 }
146
147
148 if (sourceZkCluster != null) {
149 Configuration peerConf =
150 HBaseConfiguration.createClusterConf(job.getConfiguration(), sourceZkCluster);
151 TableMapReduceUtil.initCredentialsForCluster(job, peerConf);
152 }
153 return job;
154 }
155
156 public static class SyncMapper extends TableMapper<ImmutableBytesWritable, Mutation> {
157 Path sourceHashDir;
158
159 Connection sourceConnection;
160 Connection targetConnection;
161 Table sourceTable;
162 Table targetTable;
163 boolean dryRun;
164
165 HashTable.TableHash sourceTableHash;
166 HashTable.TableHash.Reader sourceHashReader;
167 ImmutableBytesWritable currentSourceHash;
168 ImmutableBytesWritable nextSourceKey;
169 HashTable.ResultHasher targetHasher;
170
171 Throwable mapperException;
172
173 public static enum Counter {BATCHES, HASHES_MATCHED, HASHES_NOT_MATCHED, SOURCEMISSINGROWS,
174 SOURCEMISSINGCELLS, TARGETMISSINGROWS, TARGETMISSINGCELLS, ROWSWITHDIFFS, DIFFERENTCELLVALUES,
175 MATCHINGROWS, MATCHINGCELLS, EMPTY_BATCHES, RANGESMATCHED, RANGESNOTMATCHED};
176
177 @Override
178 protected void setup(Context context) throws IOException {
179
180 Configuration conf = context.getConfiguration();
181 sourceHashDir = new Path(conf.get(SOURCE_HASH_DIR_CONF_KEY));
182 sourceConnection = openConnection(conf, SOURCE_ZK_CLUSTER_CONF_KEY, null);
183 targetConnection = openConnection(conf, TARGET_ZK_CLUSTER_CONF_KEY,
184 TableOutputFormat.OUTPUT_CONF_PREFIX);
185 sourceTable = openTable(sourceConnection, conf, SOURCE_TABLE_CONF_KEY);
186 targetTable = openTable(targetConnection, conf, TARGET_TABLE_CONF_KEY);
187 dryRun = conf.getBoolean(SOURCE_TABLE_CONF_KEY, false);
188
189 sourceTableHash = HashTable.TableHash.read(conf, sourceHashDir);
190 LOG.info("Read source hash manifest: " + sourceTableHash);
191 LOG.info("Read " + sourceTableHash.partitions.size() + " partition keys");
192
193 TableSplit split = (TableSplit) context.getInputSplit();
194 ImmutableBytesWritable splitStartKey = new ImmutableBytesWritable(split.getStartRow());
195
196 sourceHashReader = sourceTableHash.newReader(conf, splitStartKey);
197 findNextKeyHashPair();
198
199
200
201
202 targetHasher = new HashTable.ResultHasher();
203 }
204
205 private static Connection openConnection(Configuration conf, String zkClusterConfKey,
206 String configPrefix)
207 throws IOException {
208 String zkCluster = conf.get(zkClusterConfKey);
209 Configuration clusterConf = HBaseConfiguration.createClusterConf(conf,
210 zkCluster, configPrefix);
211 return ConnectionFactory.createConnection(clusterConf);
212 }
213
214 private static Table openTable(Connection connection, Configuration conf,
215 String tableNameConfKey) throws IOException {
216 return connection.getTable(TableName.valueOf(conf.get(tableNameConfKey)));
217 }
218
219
220
221
222
223 private void findNextKeyHashPair() throws IOException {
224 boolean hasNext = sourceHashReader.next();
225 if (hasNext) {
226 nextSourceKey = sourceHashReader.getCurrentKey();
227 } else {
228
229 nextSourceKey = null;
230 }
231 }
232
233 @Override
234 protected void map(ImmutableBytesWritable key, Result value, Context context)
235 throws IOException, InterruptedException {
236 try {
237
238 while (nextSourceKey != null && key.compareTo(nextSourceKey) >= 0) {
239 moveToNextBatch(context);
240 }
241
242
243 if (targetHasher.isBatchStarted()) {
244 targetHasher.hashResult(value);
245 }
246 } catch (Throwable t) {
247 mapperException = t;
248 Throwables.propagateIfInstanceOf(t, IOException.class);
249 Throwables.propagateIfInstanceOf(t, InterruptedException.class);
250 Throwables.propagate(t);
251 }
252 }
253
254
255
256
257
258 private void moveToNextBatch(Context context) throws IOException, InterruptedException {
259 if (targetHasher.isBatchStarted()) {
260 finishBatchAndCompareHashes(context);
261 }
262 targetHasher.startBatch(nextSourceKey);
263 currentSourceHash = sourceHashReader.getCurrentHash();
264
265 findNextKeyHashPair();
266 }
267
268
269
270
271
272
273 private void finishBatchAndCompareHashes(Context context)
274 throws IOException, InterruptedException {
275 targetHasher.finishBatch();
276 context.getCounter(Counter.BATCHES).increment(1);
277 if (targetHasher.getBatchSize() == 0) {
278 context.getCounter(Counter.EMPTY_BATCHES).increment(1);
279 }
280 ImmutableBytesWritable targetHash = targetHasher.getBatchHash();
281 if (targetHash.equals(currentSourceHash)) {
282 context.getCounter(Counter.HASHES_MATCHED).increment(1);
283 } else {
284 context.getCounter(Counter.HASHES_NOT_MATCHED).increment(1);
285
286 ImmutableBytesWritable stopRow = nextSourceKey == null
287 ? new ImmutableBytesWritable(sourceTableHash.stopRow)
288 : nextSourceKey;
289
290 if (LOG.isDebugEnabled()) {
291 LOG.debug("Hash mismatch. Key range: " + toHex(targetHasher.getBatchStartKey())
292 + " to " + toHex(stopRow)
293 + " sourceHash: " + toHex(currentSourceHash)
294 + " targetHash: " + toHex(targetHash));
295 }
296
297 syncRange(context, targetHasher.getBatchStartKey(), stopRow);
298 }
299 }
300 private static String toHex(ImmutableBytesWritable bytes) {
301 return Bytes.toHex(bytes.get(), bytes.getOffset(), bytes.getLength());
302 }
303
304 private static final CellScanner EMPTY_CELL_SCANNER
305 = new CellScanner(Iterators.<Result>emptyIterator());
306
307
308
309
310
311
312 private void syncRange(Context context, ImmutableBytesWritable startRow,
313 ImmutableBytesWritable stopRow) throws IOException, InterruptedException {
314 Scan scan = sourceTableHash.initScan();
315 scan.setStartRow(startRow.copyBytes());
316 scan.setStopRow(stopRow.copyBytes());
317
318 ResultScanner sourceScanner = sourceTable.getScanner(scan);
319 CellScanner sourceCells = new CellScanner(sourceScanner.iterator());
320
321 ResultScanner targetScanner = targetTable.getScanner(new Scan(scan));
322 CellScanner targetCells = new CellScanner(targetScanner.iterator());
323
324 boolean rangeMatched = true;
325 byte[] nextSourceRow = sourceCells.nextRow();
326 byte[] nextTargetRow = targetCells.nextRow();
327 while(nextSourceRow != null || nextTargetRow != null) {
328 boolean rowMatched;
329 int rowComparison = compareRowKeys(nextSourceRow, nextTargetRow);
330 if (rowComparison < 0) {
331 if (LOG.isInfoEnabled()) {
332 LOG.info("Target missing row: " + Bytes.toHex(nextSourceRow));
333 }
334 context.getCounter(Counter.TARGETMISSINGROWS).increment(1);
335
336 rowMatched = syncRowCells(context, nextSourceRow, sourceCells, EMPTY_CELL_SCANNER);
337 nextSourceRow = sourceCells.nextRow();
338 } else if (rowComparison > 0) {
339 if (LOG.isInfoEnabled()) {
340 LOG.info("Source missing row: " + Bytes.toHex(nextTargetRow));
341 }
342 context.getCounter(Counter.SOURCEMISSINGROWS).increment(1);
343
344 rowMatched = syncRowCells(context, nextTargetRow, EMPTY_CELL_SCANNER, targetCells);
345 nextTargetRow = targetCells.nextRow();
346 } else {
347
348 rowMatched = syncRowCells(context, nextSourceRow, sourceCells, targetCells);
349 nextSourceRow = sourceCells.nextRow();
350 nextTargetRow = targetCells.nextRow();
351 }
352
353 if (!rowMatched) {
354 rangeMatched = false;
355 }
356 }
357
358 sourceScanner.close();
359 targetScanner.close();
360
361 context.getCounter(rangeMatched ? Counter.RANGESMATCHED : Counter.RANGESNOTMATCHED)
362 .increment(1);
363 }
364
365 private static class CellScanner {
366 private final Iterator<Result> results;
367
368 private byte[] currentRow;
369 private Result currentRowResult;
370 private int nextCellInRow;
371
372 private Result nextRowResult;
373
374 public CellScanner(Iterator<Result> results) {
375 this.results = results;
376 }
377
378
379
380
381
382 public byte[] nextRow() {
383 if (nextRowResult == null) {
384
385 while (results.hasNext()) {
386 nextRowResult = results.next();
387 Cell nextCell = nextRowResult.rawCells()[0];
388 if (currentRow == null
389 || !Bytes.equals(currentRow, 0, currentRow.length, nextCell.getRowArray(),
390 nextCell.getRowOffset(), nextCell.getRowLength())) {
391
392 break;
393 } else {
394
395 nextRowResult = null;
396 }
397 }
398
399 if (nextRowResult == null) {
400
401 currentRowResult = null;
402 currentRow = null;
403 return null;
404 }
405 }
406
407
408 currentRowResult = nextRowResult;
409 nextCellInRow = 0;
410 currentRow = currentRowResult.getRow();
411 nextRowResult = null;
412 return currentRow;
413 }
414
415
416
417
418 public Cell nextCellInRow() {
419 if (currentRowResult == null) {
420
421 return null;
422 }
423
424 Cell nextCell = currentRowResult.rawCells()[nextCellInRow];
425 nextCellInRow++;
426 if (nextCellInRow == currentRowResult.size()) {
427 if (results.hasNext()) {
428 Result result = results.next();
429 Cell cell = result.rawCells()[0];
430 if (Bytes.equals(currentRow, 0, currentRow.length, cell.getRowArray(),
431 cell.getRowOffset(), cell.getRowLength())) {
432
433 currentRowResult = result;
434 nextCellInRow = 0;
435 } else {
436
437 nextRowResult = result;
438
439 currentRowResult = null;
440 }
441 } else {
442
443 currentRowResult = null;
444 }
445 }
446 return nextCell;
447 }
448 }
449
450
451
452
453
454
455
456 private boolean syncRowCells(Context context, byte[] rowKey, CellScanner sourceCells,
457 CellScanner targetCells) throws IOException, InterruptedException {
458 Put put = null;
459 Delete delete = null;
460 long matchingCells = 0;
461 boolean matchingRow = true;
462 Cell sourceCell = sourceCells.nextCellInRow();
463 Cell targetCell = targetCells.nextCellInRow();
464 while (sourceCell != null || targetCell != null) {
465
466 int cellKeyComparison = compareCellKeysWithinRow(sourceCell, targetCell);
467 if (cellKeyComparison < 0) {
468 if (LOG.isDebugEnabled()) {
469 LOG.debug("Target missing cell: " + sourceCell);
470 }
471 context.getCounter(Counter.TARGETMISSINGCELLS).increment(1);
472 matchingRow = false;
473
474 if (!dryRun) {
475 if (put == null) {
476 put = new Put(rowKey);
477 }
478 put.add(sourceCell);
479 }
480
481 sourceCell = sourceCells.nextCellInRow();
482 } else if (cellKeyComparison > 0) {
483 if (LOG.isDebugEnabled()) {
484 LOG.debug("Source missing cell: " + targetCell);
485 }
486 context.getCounter(Counter.SOURCEMISSINGCELLS).increment(1);
487 matchingRow = false;
488
489 if (!dryRun) {
490 if (delete == null) {
491 delete = new Delete(rowKey);
492 }
493
494 delete.addColumn(CellUtil.cloneFamily(targetCell),
495 CellUtil.cloneQualifier(targetCell), targetCell.getTimestamp());
496 }
497
498 targetCell = targetCells.nextCellInRow();
499 } else {
500
501 if (CellUtil.matchingValue(sourceCell, targetCell)) {
502 matchingCells++;
503 } else {
504 if (LOG.isDebugEnabled()) {
505 LOG.debug("Different values: ");
506 LOG.debug(" source cell: " + sourceCell
507 + " value: " + Bytes.toHex(sourceCell.getValueArray(),
508 sourceCell.getValueOffset(), sourceCell.getValueLength()));
509 LOG.debug(" target cell: " + targetCell
510 + " value: " + Bytes.toHex(targetCell.getValueArray(),
511 targetCell.getValueOffset(), targetCell.getValueLength()));
512 }
513 context.getCounter(Counter.DIFFERENTCELLVALUES).increment(1);
514 matchingRow = false;
515
516 if (!dryRun) {
517
518 if (put == null) {
519 put = new Put(rowKey);
520 }
521 put.add(sourceCell);
522 }
523 }
524 sourceCell = sourceCells.nextCellInRow();
525 targetCell = targetCells.nextCellInRow();
526 }
527
528 if (!dryRun && sourceTableHash.scanBatch > 0) {
529 if (put != null && put.size() >= sourceTableHash.scanBatch) {
530 context.write(new ImmutableBytesWritable(rowKey), put);
531 put = null;
532 }
533 if (delete != null && delete.size() >= sourceTableHash.scanBatch) {
534 context.write(new ImmutableBytesWritable(rowKey), delete);
535 delete = null;
536 }
537 }
538 }
539
540 if (!dryRun) {
541 if (put != null) {
542 context.write(new ImmutableBytesWritable(rowKey), put);
543 }
544 if (delete != null) {
545 context.write(new ImmutableBytesWritable(rowKey), delete);
546 }
547 }
548
549 if (matchingCells > 0) {
550 context.getCounter(Counter.MATCHINGCELLS).increment(matchingCells);
551 }
552 if (matchingRow) {
553 context.getCounter(Counter.MATCHINGROWS).increment(1);
554 return true;
555 } else {
556 context.getCounter(Counter.ROWSWITHDIFFS).increment(1);
557 return false;
558 }
559 }
560
561 private static final CellComparator cellComparator = new CellComparator();
562
563
564
565
566 private static int compareRowKeys(byte[] r1, byte[] r2) {
567 if (r1 == null) {
568 return 1;
569 } else if (r2 == null) {
570 return -1;
571 } else {
572 return cellComparator.compareRows(r1, 0, r1.length, r2, 0, r2.length);
573 }
574 }
575
576
577
578
579
580
581 private static int compareCellKeysWithinRow(Cell c1, Cell c2) {
582 if (c1 == null) {
583 return 1;
584 }
585 if (c2 == null) {
586 return -1;
587 }
588
589 int result = CellComparator.compareFamilies(c1, c2);
590 if (result != 0) {
591 return result;
592 }
593
594 result = CellComparator.compareQualifiers(c1, c2);
595 if (result != 0) {
596 return result;
597 }
598
599
600 return CellComparator.compareTimestamps(c1, c2);
601 }
602
603 @Override
604 protected void cleanup(Context context)
605 throws IOException, InterruptedException {
606 if (mapperException == null) {
607 try {
608 finishRemainingHashRanges(context);
609 } catch (Throwable t) {
610 mapperException = t;
611 }
612 }
613
614 try {
615 sourceTable.close();
616 targetTable.close();
617 sourceConnection.close();
618 targetConnection.close();
619 } catch (Throwable t) {
620 if (mapperException == null) {
621 mapperException = t;
622 } else {
623 LOG.error("Suppressing exception from closing tables", t);
624 }
625 }
626
627
628 if (mapperException != null) {
629 Throwables.propagateIfInstanceOf(mapperException, IOException.class);
630 Throwables.propagateIfInstanceOf(mapperException, InterruptedException.class);
631 Throwables.propagate(mapperException);
632 }
633 }
634
635 private void finishRemainingHashRanges(Context context) throws IOException,
636 InterruptedException {
637 TableSplit split = (TableSplit) context.getInputSplit();
638 byte[] splitEndRow = split.getEndRow();
639 boolean reachedEndOfTable = HashTable.isTableEndRow(splitEndRow);
640
641
642 while (nextSourceKey != null
643 && (nextSourceKey.compareTo(splitEndRow) < 0 || reachedEndOfTable)) {
644 moveToNextBatch(context);
645 }
646
647 if (targetHasher.isBatchStarted()) {
648
649
650 if ((nextSourceKey != null && nextSourceKey.compareTo(splitEndRow) > 0)
651 || (nextSourceKey == null && !Bytes.equals(splitEndRow, sourceTableHash.stopRow))) {
652
653
654 Scan scan = sourceTableHash.initScan();
655 scan.setStartRow(splitEndRow);
656 if (nextSourceKey == null) {
657 scan.setStopRow(sourceTableHash.stopRow);
658 } else {
659 scan.setStopRow(nextSourceKey.copyBytes());
660 }
661
662 ResultScanner targetScanner = null;
663 try {
664 targetScanner = targetTable.getScanner(scan);
665 for (Result row : targetScanner) {
666 targetHasher.hashResult(row);
667 }
668 } finally {
669 if (targetScanner != null) {
670 targetScanner.close();
671 }
672 }
673 }
674
675 finishBatchAndCompareHashes(context);
676 }
677 }
678 }
679
680 private static final int NUM_ARGS = 3;
681 private static void printUsage(final String errorMsg) {
682 if (errorMsg != null && errorMsg.length() > 0) {
683 System.err.println("ERROR: " + errorMsg);
684 System.err.println();
685 }
686 System.err.println("Usage: SyncTable [options] <sourcehashdir> <sourcetable> <targettable>");
687 System.err.println();
688 System.err.println("Options:");
689
690 System.err.println(" sourcezkcluster ZK cluster key of the source table");
691 System.err.println(" (defaults to cluster in classpath's config)");
692 System.err.println(" targetzkcluster ZK cluster key of the target table");
693 System.err.println(" (defaults to cluster in classpath's config)");
694 System.err.println(" dryrun if true, output counters but no writes");
695 System.err.println(" (defaults to false)");
696 System.err.println();
697 System.err.println("Args:");
698 System.err.println(" sourcehashdir path to HashTable output dir for source table");
699 System.err.println(" (see org.apache.hadoop.hbase.mapreduce.HashTable)");
700 System.err.println(" sourcetable Name of the source table to sync from");
701 System.err.println(" targettable Name of the target table to sync to");
702 System.err.println();
703 System.err.println("Examples:");
704 System.err.println(" For a dry run SyncTable of tableA from a remote source cluster");
705 System.err.println(" to a local target cluster:");
706 System.err.println(" $ bin/hbase " +
707 "org.apache.hadoop.hbase.mapreduce.SyncTable --dryrun=true"
708 + " --sourcezkcluster=zk1.example.com,zk2.example.com,zk3.example.com:2181:/hbase"
709 + " hdfs://nn:9000/hashes/tableA tableA tableA");
710 }
711
712 private boolean doCommandLine(final String[] args) {
713 if (args.length < NUM_ARGS) {
714 printUsage(null);
715 return false;
716 }
717 try {
718 sourceHashDir = new Path(args[args.length - 3]);
719 sourceTableName = args[args.length - 2];
720 targetTableName = args[args.length - 1];
721
722 for (int i = 0; i < args.length - NUM_ARGS; i++) {
723 String cmd = args[i];
724 if (cmd.equals("-h") || cmd.startsWith("--h")) {
725 printUsage(null);
726 return false;
727 }
728
729 final String sourceZkClusterKey = "--sourcezkcluster=";
730 if (cmd.startsWith(sourceZkClusterKey)) {
731 sourceZkCluster = cmd.substring(sourceZkClusterKey.length());
732 continue;
733 }
734
735 final String targetZkClusterKey = "--targetzkcluster=";
736 if (cmd.startsWith(targetZkClusterKey)) {
737 targetZkCluster = cmd.substring(targetZkClusterKey.length());
738 continue;
739 }
740
741 final String dryRunKey = "--dryrun=";
742 if (cmd.startsWith(dryRunKey)) {
743 dryRun = Boolean.parseBoolean(cmd.substring(dryRunKey.length()));
744 continue;
745 }
746
747 printUsage("Invalid argument '" + cmd + "'");
748 return false;
749 }
750
751
752 } catch (Exception e) {
753 e.printStackTrace();
754 printUsage("Can't start because " + e.getMessage());
755 return false;
756 }
757 return true;
758 }
759
760
761
762
763 public static void main(String[] args) throws Exception {
764 int ret = ToolRunner.run(new SyncTable(HBaseConfiguration.create()), args);
765 System.exit(ret);
766 }
767
768 @Override
769 public int run(String[] args) throws Exception {
770 String[] otherArgs = new GenericOptionsParser(getConf(), args).getRemainingArgs();
771 if (!doCommandLine(otherArgs)) {
772 return 1;
773 }
774
775 Job job = createSubmittableJob(otherArgs);
776 if (!job.waitForCompletion(true)) {
777 LOG.info("Map-reduce job failed!");
778 return 1;
779 }
780 counters = job.getCounters();
781 return 0;
782 }
783
784 }