1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.util;
20
21 import java.io.IOException;
22 import java.math.BigInteger;
23 import java.util.Arrays;
24 import java.util.Collection;
25 import java.util.Collections;
26 import java.util.Comparator;
27 import java.util.LinkedList;
28 import java.util.List;
29 import java.util.Set;
30 import java.util.TreeMap;
31
32 import org.apache.commons.cli.CommandLine;
33 import org.apache.commons.cli.GnuParser;
34 import org.apache.commons.cli.HelpFormatter;
35 import org.apache.commons.cli.OptionBuilder;
36 import org.apache.commons.cli.Options;
37 import org.apache.commons.cli.ParseException;
38 import org.apache.commons.lang.ArrayUtils;
39 import org.apache.commons.lang.StringUtils;
40 import org.apache.commons.logging.Log;
41 import org.apache.commons.logging.LogFactory;
42 import org.apache.hadoop.conf.Configuration;
43 import org.apache.hadoop.fs.FSDataInputStream;
44 import org.apache.hadoop.fs.FSDataOutputStream;
45 import org.apache.hadoop.fs.FileSystem;
46 import org.apache.hadoop.fs.Path;
47 import org.apache.hadoop.hbase.ClusterStatus;
48 import org.apache.hadoop.hbase.HBaseConfiguration;
49 import org.apache.hadoop.hbase.HConstants;
50 import org.apache.hadoop.hbase.HColumnDescriptor;
51 import org.apache.hadoop.hbase.HRegionInfo;
52 import org.apache.hadoop.hbase.HRegionLocation;
53 import org.apache.hadoop.hbase.HTableDescriptor;
54 import org.apache.hadoop.hbase.MetaTableAccessor;
55 import org.apache.hadoop.hbase.ServerName;
56 import org.apache.hadoop.hbase.TableName;
57 import org.apache.hadoop.hbase.classification.InterfaceAudience;
58 import org.apache.hadoop.hbase.client.Admin;
59 import org.apache.hadoop.hbase.client.ClusterConnection;
60 import org.apache.hadoop.hbase.client.Connection;
61 import org.apache.hadoop.hbase.client.ConnectionFactory;
62 import org.apache.hadoop.hbase.client.NoServerForRegionException;
63 import org.apache.hadoop.hbase.client.RegionLocator;
64 import org.apache.hadoop.hbase.client.Table;
65 import org.apache.hadoop.hbase.regionserver.HRegionFileSystem;
66
67 import com.google.common.base.Preconditions;
68 import com.google.common.collect.Lists;
69 import com.google.common.collect.Maps;
70 import com.google.common.collect.Sets;
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143 @InterfaceAudience.Private
144 public class RegionSplitter {
145 private static final Log LOG = LogFactory.getLog(RegionSplitter.class);
146
147
148
149
150
151
152
153
154
155
156 public interface SplitAlgorithm {
157
158
159
160
161
162
163
164
165
166 byte[] split(byte[] start, byte[] end);
167
168
169
170
171
172
173
174
175
176
177
178
179
180 byte[][] split(int numRegions);
181
182
183
184
185
186
187
188
189
190
191 byte[][] split(byte[] start, byte[] end, int numSplits, boolean inclusive);
192
193
194
195
196
197
198
199
200 byte[] firstRow();
201
202
203
204
205
206
207
208
209 byte[] lastRow();
210
211
212
213
214
215
216
217
218
219 void setFirstRow(String userInput);
220
221
222
223
224
225
226
227
228
229
230 void setLastRow(String userInput);
231
232
233
234
235
236
237 byte[] strToRow(String input);
238
239
240
241
242
243
244 String rowToStr(byte[] row);
245
246
247
248
249 String separator();
250
251
252
253
254
255 void setFirstRow(byte[] userInput);
256
257
258
259
260
261 void setLastRow(byte[] userInput);
262 }
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299 @SuppressWarnings("static-access")
300 public static void main(String[] args) throws IOException,
301 InterruptedException, ParseException {
302 Configuration conf = HBaseConfiguration.create();
303
304
305 Options opt = new Options();
306 opt.addOption(OptionBuilder.withArgName("property=value").hasArg()
307 .withDescription("Override HBase Configuration Settings").create("D"));
308 opt.addOption(OptionBuilder.withArgName("region count").hasArg()
309 .withDescription(
310 "Create a new table with a pre-split number of regions")
311 .create("c"));
312 opt.addOption(OptionBuilder.withArgName("family:family:...").hasArg()
313 .withDescription(
314 "Column Families to create with new table. Required with -c")
315 .create("f"));
316 opt.addOption("h", false, "Print this usage help");
317 opt.addOption("r", false, "Perform a rolling split of an existing region");
318 opt.addOption(OptionBuilder.withArgName("count").hasArg().withDescription(
319 "Max outstanding splits that have unfinished major compactions")
320 .create("o"));
321 opt.addOption(null, "firstrow", true,
322 "First Row in Table for Split Algorithm");
323 opt.addOption(null, "lastrow", true,
324 "Last Row in Table for Split Algorithm");
325 opt.addOption(null, "risky", false,
326 "Skip verification steps to complete quickly."
327 + "STRONGLY DISCOURAGED for production systems. ");
328 CommandLine cmd = new GnuParser().parse(opt, args);
329
330 if (cmd.hasOption("D")) {
331 for (String confOpt : cmd.getOptionValues("D")) {
332 String[] kv = confOpt.split("=", 2);
333 if (kv.length == 2) {
334 conf.set(kv[0], kv[1]);
335 LOG.debug("-D configuration override: " + kv[0] + "=" + kv[1]);
336 } else {
337 throw new ParseException("-D option format invalid: " + confOpt);
338 }
339 }
340 }
341
342 if (cmd.hasOption("risky")) {
343 conf.setBoolean("split.verify", false);
344 }
345
346 boolean createTable = cmd.hasOption("c") && cmd.hasOption("f");
347 boolean rollingSplit = cmd.hasOption("r");
348 boolean oneOperOnly = createTable ^ rollingSplit;
349
350 if (2 != cmd.getArgList().size() || !oneOperOnly || cmd.hasOption("h")) {
351 new HelpFormatter().printHelp("RegionSplitter <TABLE> <SPLITALGORITHM>\n"+
352 "SPLITALGORITHM is a java class name of a class implementing " +
353 "SplitAlgorithm, or one of the special strings HexStringSplit " +
354 "or UniformSplit, which are built-in split algorithms. " +
355 "HexStringSplit treats keys as hexadecimal ASCII, and " +
356 "UniformSplit treats keys as arbitrary bytes.", opt);
357 return;
358 }
359 TableName tableName = TableName.valueOf(cmd.getArgs()[0]);
360 String splitClass = cmd.getArgs()[1];
361 SplitAlgorithm splitAlgo = newSplitAlgoInstance(conf, splitClass);
362
363 if (cmd.hasOption("firstrow")) {
364 splitAlgo.setFirstRow(cmd.getOptionValue("firstrow"));
365 }
366 if (cmd.hasOption("lastrow")) {
367 splitAlgo.setLastRow(cmd.getOptionValue("lastrow"));
368 }
369
370 if (createTable) {
371 conf.set("split.count", cmd.getOptionValue("c"));
372 createPresplitTable(tableName, splitAlgo, cmd.getOptionValue("f").split(":"), conf);
373 }
374
375 if (rollingSplit) {
376 if (cmd.hasOption("o")) {
377 conf.set("split.outstanding", cmd.getOptionValue("o"));
378 }
379 rollingSplit(tableName, splitAlgo, conf);
380 }
381 }
382
383 static void createPresplitTable(TableName tableName, SplitAlgorithm splitAlgo,
384 String[] columnFamilies, Configuration conf)
385 throws IOException, InterruptedException {
386 final int splitCount = conf.getInt("split.count", 0);
387 Preconditions.checkArgument(splitCount > 1, "Split count must be > 1");
388
389 Preconditions.checkArgument(columnFamilies.length > 0,
390 "Must specify at least one column family. ");
391 LOG.debug("Creating table " + tableName + " with " + columnFamilies.length
392 + " column families. Presplitting to " + splitCount + " regions");
393
394 HTableDescriptor desc = new HTableDescriptor(tableName);
395 for (String cf : columnFamilies) {
396 desc.addFamily(new HColumnDescriptor(Bytes.toBytes(cf)));
397 }
398 try (Connection connection = ConnectionFactory.createConnection(conf)) {
399 Admin admin = connection.getAdmin();
400 try {
401 Preconditions.checkArgument(!admin.tableExists(tableName),
402 "Table already exists: " + tableName);
403 admin.createTable(desc, splitAlgo.split(splitCount));
404 } finally {
405 admin.close();
406 }
407 LOG.debug("Table created! Waiting for regions to show online in META...");
408 if (!conf.getBoolean("split.verify", true)) {
409
410 int onlineRegions = 0;
411 while (onlineRegions < splitCount) {
412 onlineRegions = MetaTableAccessor.getRegionCount(connection, tableName);
413 LOG.debug(onlineRegions + " of " + splitCount + " regions online...");
414 if (onlineRegions < splitCount) {
415 Thread.sleep(10 * 1000);
416 }
417 }
418 }
419 LOG.debug("Finished creating table with " + splitCount + " regions");
420 }
421 }
422
423
424
425
426
427
428
429 private static int getRegionServerCount(final Connection connection) throws IOException {
430 try (Admin admin = connection.getAdmin()) {
431 ClusterStatus status = admin.getClusterStatus();
432 Collection<ServerName> servers = status.getServers();
433 return servers == null || servers.isEmpty()? 0: servers.size();
434 }
435 }
436
437 private static byte [] readFile(final FileSystem fs, final Path path) throws IOException {
438 FSDataInputStream tmpIn = fs.open(path);
439 try {
440 byte [] rawData = new byte[tmpIn.available()];
441 tmpIn.readFully(rawData);
442 return rawData;
443 } finally {
444 tmpIn.close();
445 }
446 }
447
448 static void rollingSplit(TableName tableName, SplitAlgorithm splitAlgo, Configuration conf)
449 throws IOException, InterruptedException {
450 final int minOS = conf.getInt("split.outstanding", 2);
451 try (Connection connection = ConnectionFactory.createConnection(conf)) {
452
453 final int MAX_OUTSTANDING = Math.max(getRegionServerCount(connection) / 2, minOS);
454
455 Path hbDir = FSUtils.getRootDir(conf);
456 Path tableDir = FSUtils.getTableDir(hbDir, tableName);
457 Path splitFile = new Path(tableDir, "_balancedSplit");
458 FileSystem fs = FileSystem.get(conf);
459
460
461 LinkedList<Pair<byte[], byte[]>> tmpRegionSet = null;
462 try (Table table = connection.getTable(tableName)) {
463 tmpRegionSet = getSplits(connection, tableName, splitAlgo);
464 }
465 LinkedList<Pair<byte[], byte[]>> outstanding = Lists.newLinkedList();
466 int splitCount = 0;
467 final int origCount = tmpRegionSet.size();
468
469
470
471
472 LOG.debug("Bucketing regions by regionserver...");
473 TreeMap<String, LinkedList<Pair<byte[], byte[]>>> daughterRegions =
474 Maps.newTreeMap();
475
476 try (RegionLocator regionLocator = connection.getRegionLocator(tableName)) {
477 for (Pair<byte[], byte[]> dr : tmpRegionSet) {
478 String rsLocation = regionLocator.getRegionLocation(dr.getSecond()).getHostnamePort();
479 if (!daughterRegions.containsKey(rsLocation)) {
480 LinkedList<Pair<byte[], byte[]>> entry = Lists.newLinkedList();
481 daughterRegions.put(rsLocation, entry);
482 }
483 daughterRegions.get(rsLocation).add(dr);
484 }
485 LOG.debug("Done with bucketing. Split time!");
486 long startTime = System.currentTimeMillis();
487
488
489 byte[] rawData = readFile(fs, splitFile);
490
491 FSDataOutputStream splitOut = fs.create(splitFile);
492 try {
493 splitOut.write(rawData);
494
495 try {
496
497 while (!daughterRegions.isEmpty()) {
498 LOG.debug(daughterRegions.size() + " RS have regions to splt.");
499
500
501 final TreeMap<ServerName, Integer> rsSizes = Maps.newTreeMap();
502 List<HRegionLocation> hrls = regionLocator.getAllRegionLocations();
503 for (HRegionLocation hrl: hrls) {
504 ServerName sn = hrl.getServerName();
505 if (rsSizes.containsKey(sn)) {
506 rsSizes.put(sn, rsSizes.get(sn) + 1);
507 } else {
508 rsSizes.put(sn, 1);
509 }
510 }
511
512
513 List<String> serversLeft = Lists.newArrayList(daughterRegions .keySet());
514 Collections.sort(serversLeft, new Comparator<String>() {
515 public int compare(String o1, String o2) {
516 return rsSizes.get(o1).compareTo(rsSizes.get(o2));
517 }
518 });
519
520
521
522 for (String rsLoc : serversLeft) {
523 Pair<byte[], byte[]> dr = null;
524
525
526 LOG.debug("Finding a region on " + rsLoc);
527 LinkedList<Pair<byte[], byte[]>> regionList = daughterRegions.get(rsLoc);
528 while (!regionList.isEmpty()) {
529 dr = regionList.pop();
530
531
532 byte[] split = dr.getSecond();
533 HRegionLocation regionLoc = regionLocator.getRegionLocation(split);
534
535
536 String newRs = regionLoc.getHostnamePort();
537 if (newRs.compareTo(rsLoc) != 0) {
538 LOG.debug("Region with " + splitAlgo.rowToStr(split)
539 + " moved to " + newRs + ". Relocating...");
540
541 if (!daughterRegions.containsKey(newRs)) {
542 LinkedList<Pair<byte[], byte[]>> entry = Lists.newLinkedList();
543 daughterRegions.put(newRs, entry);
544 }
545 daughterRegions.get(newRs).add(dr);
546 dr = null;
547 continue;
548 }
549
550
551 byte[] sk = regionLoc.getRegionInfo().getStartKey();
552 if (sk.length != 0) {
553 if (Bytes.equals(split, sk)) {
554 LOG.debug("Region already split on "
555 + splitAlgo.rowToStr(split) + ". Skipping this region...");
556 ++splitCount;
557 dr = null;
558 continue;
559 }
560 byte[] start = dr.getFirst();
561 Preconditions.checkArgument(Bytes.equals(start, sk), splitAlgo
562 .rowToStr(start) + " != " + splitAlgo.rowToStr(sk));
563 }
564
565
566 break;
567 }
568 if (regionList.isEmpty()) {
569 daughterRegions.remove(rsLoc);
570 }
571 if (dr == null)
572 continue;
573
574
575 byte[] split = dr.getSecond();
576 LOG.debug("Splitting at " + splitAlgo.rowToStr(split));
577 try (Admin admin = connection.getAdmin()) {
578 admin.split(tableName, split);
579 }
580
581 LinkedList<Pair<byte[], byte[]>> finished = Lists.newLinkedList();
582 LinkedList<Pair<byte[], byte[]>> local_finished = Lists.newLinkedList();
583 if (conf.getBoolean("split.verify", true)) {
584
585 outstanding.addLast(dr);
586
587 while (outstanding.size() >= MAX_OUTSTANDING) {
588 LOG.debug("Wait for outstanding splits " + outstanding.size());
589 local_finished = splitScan(outstanding, connection, tableName, splitAlgo);
590 if (local_finished.isEmpty()) {
591 Thread.sleep(30 * 1000);
592 } else {
593 finished.addAll(local_finished);
594 outstanding.removeAll(local_finished);
595 LOG.debug(local_finished.size() + " outstanding splits finished");
596 }
597 }
598 } else {
599 finished.add(dr);
600 }
601
602
603 for (Pair<byte[], byte[]> region : finished) {
604 splitOut.writeChars("- " + splitAlgo.rowToStr(region.getFirst())
605 + " " + splitAlgo.rowToStr(region.getSecond()) + "\n");
606 splitCount++;
607 if (splitCount % 10 == 0) {
608 long tDiff = (System.currentTimeMillis() - startTime)
609 / splitCount;
610 LOG.debug("STATUS UPDATE: " + splitCount + " / " + origCount
611 + ". Avg Time / Split = "
612 + org.apache.hadoop.util.StringUtils.formatTime(tDiff));
613 }
614 }
615 }
616 }
617 if (conf.getBoolean("split.verify", true)) {
618 while (!outstanding.isEmpty()) {
619 LOG.debug("Finally Wait for outstanding splits " + outstanding.size());
620 LinkedList<Pair<byte[], byte[]>> finished = splitScan(outstanding,
621 connection, tableName, splitAlgo);
622 if (finished.isEmpty()) {
623 Thread.sleep(30 * 1000);
624 } else {
625 outstanding.removeAll(finished);
626 for (Pair<byte[], byte[]> region : finished) {
627 splitOut.writeChars("- " + splitAlgo.rowToStr(region.getFirst())
628 + " " + splitAlgo.rowToStr(region.getSecond()) + "\n");
629 splitCount++;
630 }
631 LOG.debug("Finally " + finished.size() + " outstanding splits finished");
632 }
633 }
634 }
635 LOG.debug("All regions have been successfully split!");
636 } finally {
637 long tDiff = System.currentTimeMillis() - startTime;
638 LOG.debug("TOTAL TIME = "
639 + org.apache.hadoop.util.StringUtils.formatTime(tDiff));
640 LOG.debug("Splits = " + splitCount);
641 if (0 < splitCount) {
642 LOG.debug("Avg Time / Split = "
643 + org.apache.hadoop.util.StringUtils.formatTime(tDiff / splitCount));
644 }
645 }
646 } finally {
647 splitOut.close();
648 fs.delete(splitFile, false);
649 }
650 }
651 }
652 }
653
654
655
656
657
658 public static SplitAlgorithm newSplitAlgoInstance(Configuration conf,
659 String splitClassName) throws IOException {
660 Class<?> splitClass;
661
662
663
664 if(splitClassName.equals(HexStringSplit.class.getSimpleName())) {
665 splitClass = HexStringSplit.class;
666 } else if (splitClassName.equals(UniformSplit.class.getSimpleName())) {
667 splitClass = UniformSplit.class;
668 } else {
669 try {
670 splitClass = conf.getClassByName(splitClassName);
671 } catch (ClassNotFoundException e) {
672 throw new IOException("Couldn't load split class " + splitClassName, e);
673 }
674 if(splitClass == null) {
675 throw new IOException("Failed loading split class " + splitClassName);
676 }
677 if(!SplitAlgorithm.class.isAssignableFrom(splitClass)) {
678 throw new IOException(
679 "Specified split class doesn't implement SplitAlgorithm");
680 }
681 }
682 try {
683 return splitClass.asSubclass(SplitAlgorithm.class).newInstance();
684 } catch (Exception e) {
685 throw new IOException("Problem loading split algorithm: ", e);
686 }
687 }
688
689 static LinkedList<Pair<byte[], byte[]>> splitScan(
690 LinkedList<Pair<byte[], byte[]>> regionList,
691 final Connection connection,
692 final TableName tableName,
693 SplitAlgorithm splitAlgo)
694 throws IOException, InterruptedException {
695 LinkedList<Pair<byte[], byte[]>> finished = Lists.newLinkedList();
696 LinkedList<Pair<byte[], byte[]>> logicalSplitting = Lists.newLinkedList();
697 LinkedList<Pair<byte[], byte[]>> physicalSplitting = Lists.newLinkedList();
698
699
700 Pair<Path, Path> tableDirAndSplitFile =
701 getTableDirAndSplitFile(connection.getConfiguration(), tableName);
702 Path tableDir = tableDirAndSplitFile.getFirst();
703 FileSystem fs = tableDir.getFileSystem(connection.getConfiguration());
704
705 ((ClusterConnection)connection).clearRegionCache();
706 HTableDescriptor htd = null;
707 try (Table table = connection.getTable(tableName)) {
708 htd = table.getTableDescriptor();
709 }
710 try (RegionLocator regionLocator = connection.getRegionLocator(tableName)) {
711
712
713 for (Pair<byte[], byte[]> region : regionList) {
714 byte[] start = region.getFirst();
715 byte[] split = region.getSecond();
716
717
718 try {
719 HRegionInfo dri = regionLocator.getRegionLocation(split).getRegionInfo();
720 if (dri.isOffline() || !Bytes.equals(dri.getStartKey(), split)) {
721 logicalSplitting.add(region);
722 continue;
723 }
724 } catch (NoServerForRegionException nsfre) {
725
726 LOG.info(nsfre);
727 logicalSplitting.add(region);
728 continue;
729 }
730
731 try {
732
733
734 LinkedList<HRegionInfo> check = Lists.newLinkedList();
735 check.add(regionLocator.getRegionLocation(start).getRegionInfo());
736 check.add(regionLocator.getRegionLocation(split).getRegionInfo());
737 for (HRegionInfo hri : check.toArray(new HRegionInfo[check.size()])) {
738 byte[] sk = hri.getStartKey();
739 if (sk.length == 0)
740 sk = splitAlgo.firstRow();
741
742 HRegionFileSystem regionFs = HRegionFileSystem.openRegionFromFileSystem(
743 connection.getConfiguration(), fs, tableDir, hri, true);
744
745
746 boolean refFound = false;
747 for (HColumnDescriptor c : htd.getFamilies()) {
748 if ((refFound = regionFs.hasReferences(c.getNameAsString()))) {
749 break;
750 }
751 }
752
753
754 if (!refFound) {
755 check.remove(hri);
756 }
757 }
758 if (check.isEmpty()) {
759 finished.add(region);
760 } else {
761 physicalSplitting.add(region);
762 }
763 } catch (NoServerForRegionException nsfre) {
764 LOG.debug("No Server Exception thrown for: " + splitAlgo.rowToStr(start));
765 physicalSplitting.add(region);
766 ((ClusterConnection)connection).clearRegionCache();
767 }
768 }
769
770 LOG.debug("Split Scan: " + finished.size() + " finished / "
771 + logicalSplitting.size() + " split wait / "
772 + physicalSplitting.size() + " reference wait");
773
774 return finished;
775 }
776 }
777
778
779
780
781
782
783
784 private static Pair<Path, Path> getTableDirAndSplitFile(final Configuration conf,
785 final TableName tableName)
786 throws IOException {
787 Path hbDir = FSUtils.getRootDir(conf);
788 Path tableDir = FSUtils.getTableDir(hbDir, tableName);
789 Path splitFile = new Path(tableDir, "_balancedSplit");
790 return new Pair<Path, Path>(tableDir, splitFile);
791 }
792
793 static LinkedList<Pair<byte[], byte[]>> getSplits(final Connection connection,
794 TableName tableName, SplitAlgorithm splitAlgo)
795 throws IOException {
796 Pair<Path, Path> tableDirAndSplitFile =
797 getTableDirAndSplitFile(connection.getConfiguration(), tableName);
798 Path tableDir = tableDirAndSplitFile.getFirst();
799 Path splitFile = tableDirAndSplitFile.getSecond();
800
801 FileSystem fs = tableDir.getFileSystem(connection.getConfiguration());
802
803
804 Set<Pair<String, String>> daughterRegions = Sets.newHashSet();
805
806
807 if (!fs.exists(splitFile)) {
808
809 LOG.debug("No " + splitFile.getName() + " file. Calculating splits ");
810
811
812 Set<Pair<byte[], byte[]>> rows = Sets.newHashSet();
813 Pair<byte[][], byte[][]> tmp = null;
814 try (RegionLocator regionLocator = connection.getRegionLocator(tableName)) {
815 tmp = regionLocator.getStartEndKeys();
816 }
817 Preconditions.checkArgument(tmp.getFirst().length == tmp.getSecond().length,
818 "Start and End rows should be equivalent");
819 for (int i = 0; i < tmp.getFirst().length; ++i) {
820 byte[] start = tmp.getFirst()[i], end = tmp.getSecond()[i];
821 if (start.length == 0)
822 start = splitAlgo.firstRow();
823 if (end.length == 0)
824 end = splitAlgo.lastRow();
825 rows.add(Pair.newPair(start, end));
826 }
827 LOG.debug("Table " + tableName + " has " + rows.size() + " regions that will be split.");
828
829
830 Path tmpFile = new Path(tableDir, "_balancedSplit_prepare");
831 FSDataOutputStream tmpOut = fs.create(tmpFile);
832
833
834 for (Pair<byte[], byte[]> r : rows) {
835 byte[] splitPoint = splitAlgo.split(r.getFirst(), r.getSecond());
836 String startStr = splitAlgo.rowToStr(r.getFirst());
837 String splitStr = splitAlgo.rowToStr(splitPoint);
838 daughterRegions.add(Pair.newPair(startStr, splitStr));
839 LOG.debug("Will Split [" + startStr + " , "
840 + splitAlgo.rowToStr(r.getSecond()) + ") at " + splitStr);
841 tmpOut.writeChars("+ " + startStr + splitAlgo.separator() + splitStr
842 + "\n");
843 }
844 tmpOut.close();
845 fs.rename(tmpFile, splitFile);
846 } else {
847 LOG.debug("_balancedSplit file found. Replay log to restore state...");
848 FSUtils.getInstance(fs, connection.getConfiguration())
849 .recoverFileLease(fs, splitFile, connection.getConfiguration(), null);
850
851
852 FSDataInputStream tmpIn = fs.open(splitFile);
853 StringBuilder sb = new StringBuilder(tmpIn.available());
854 while (tmpIn.available() > 0) {
855 sb.append(tmpIn.readChar());
856 }
857 tmpIn.close();
858 for (String line : sb.toString().split("\n")) {
859 String[] cmd = line.split(splitAlgo.separator());
860 Preconditions.checkArgument(3 == cmd.length);
861 byte[] start = splitAlgo.strToRow(cmd[1]);
862 String startStr = splitAlgo.rowToStr(start);
863 byte[] splitPoint = splitAlgo.strToRow(cmd[2]);
864 String splitStr = splitAlgo.rowToStr(splitPoint);
865 Pair<String, String> r = Pair.newPair(startStr, splitStr);
866 if (cmd[0].equals("+")) {
867 LOG.debug("Adding: " + r);
868 daughterRegions.add(r);
869 } else {
870 LOG.debug("Removing: " + r);
871 Preconditions.checkArgument(cmd[0].equals("-"),
872 "Unknown option: " + cmd[0]);
873 Preconditions.checkState(daughterRegions.contains(r),
874 "Missing row: " + r);
875 daughterRegions.remove(r);
876 }
877 }
878 LOG.debug("Done reading. " + daughterRegions.size() + " regions left.");
879 }
880 LinkedList<Pair<byte[], byte[]>> ret = Lists.newLinkedList();
881 for (Pair<String, String> r : daughterRegions) {
882 ret.add(Pair.newPair(splitAlgo.strToRow(r.getFirst()), splitAlgo
883 .strToRow(r.getSecond())));
884 }
885 return ret;
886 }
887
888
889
890
891
892
893
894
895
896
897
898
899 public static class HexStringSplit implements SplitAlgorithm {
900 final static String DEFAULT_MIN_HEX = "00000000";
901 final static String DEFAULT_MAX_HEX = "FFFFFFFF";
902
903 String firstRow = DEFAULT_MIN_HEX;
904 BigInteger firstRowInt = BigInteger.ZERO;
905 String lastRow = DEFAULT_MAX_HEX;
906 BigInteger lastRowInt = new BigInteger(lastRow, 16);
907 int rowComparisonLength = lastRow.length();
908
909 public byte[] split(byte[] start, byte[] end) {
910 BigInteger s = convertToBigInteger(start);
911 BigInteger e = convertToBigInteger(end);
912 Preconditions.checkArgument(!e.equals(BigInteger.ZERO));
913 return convertToByte(split2(s, e));
914 }
915
916 public byte[][] split(int n) {
917 Preconditions.checkArgument(lastRowInt.compareTo(firstRowInt) > 0,
918 "last row (%s) is configured less than first row (%s)", lastRow,
919 firstRow);
920
921 BigInteger range = lastRowInt.subtract(firstRowInt).add(BigInteger.ONE);
922 Preconditions.checkState(range.compareTo(BigInteger.valueOf(n)) >= 0,
923 "split granularity (%s) is greater than the range (%s)", n, range);
924
925 BigInteger[] splits = new BigInteger[n - 1];
926 BigInteger sizeOfEachSplit = range.divide(BigInteger.valueOf(n));
927 for (int i = 1; i < n; i++) {
928
929
930 splits[i - 1] = firstRowInt.add(sizeOfEachSplit.multiply(BigInteger
931 .valueOf(i)));
932 }
933 return convertToBytes(splits);
934 }
935
936 @Override
937 public byte[][] split(byte[] start, byte[] end, int numSplits, boolean inclusive) {
938 BigInteger s = convertToBigInteger(start);
939 BigInteger e = convertToBigInteger(end);
940
941 Preconditions.checkArgument(e.compareTo(s) > 0,
942 "last row (%s) is configured less than first row (%s)", rowToStr(end),
943 end);
944
945 BigInteger range = e.subtract(s).add(BigInteger.ONE);
946 Preconditions.checkState(range.compareTo(BigInteger.valueOf(numSplits)) >= 0,
947 "split granularity (%s) is greater than the range (%s)", numSplits, range);
948
949 BigInteger[] splits = new BigInteger[numSplits - 1];
950 BigInteger sizeOfEachSplit = range.divide(BigInteger.valueOf(numSplits));
951 for (int i = 1; i < numSplits; i++) {
952
953
954 splits[i - 1] = s.add(sizeOfEachSplit.multiply(BigInteger
955 .valueOf(i)));
956 }
957
958 if (inclusive) {
959 BigInteger[] inclusiveSplitPoints = new BigInteger[numSplits + 1];
960 inclusiveSplitPoints[0] = convertToBigInteger(start);
961 inclusiveSplitPoints[numSplits] = convertToBigInteger(end);
962 System.arraycopy(splits, 0, inclusiveSplitPoints, 1, splits.length);
963 return convertToBytes(inclusiveSplitPoints);
964 } else {
965 return convertToBytes(splits);
966 }
967 }
968
969 public byte[] firstRow() {
970 return convertToByte(firstRowInt);
971 }
972
973 public byte[] lastRow() {
974 return convertToByte(lastRowInt);
975 }
976
977 public void setFirstRow(String userInput) {
978 firstRow = userInput;
979 firstRowInt = new BigInteger(firstRow, 16);
980 }
981
982 public void setLastRow(String userInput) {
983 lastRow = userInput;
984 lastRowInt = new BigInteger(lastRow, 16);
985
986 rowComparisonLength = lastRow.length();
987 }
988
989 public byte[] strToRow(String in) {
990 return convertToByte(new BigInteger(in, 16));
991 }
992
993 public String rowToStr(byte[] row) {
994 return Bytes.toStringBinary(row);
995 }
996
997 public String separator() {
998 return " ";
999 }
1000
1001 @Override
1002 public void setFirstRow(byte[] userInput) {
1003 firstRow = Bytes.toString(userInput);
1004 }
1005
1006 @Override
1007 public void setLastRow(byte[] userInput) {
1008 lastRow = Bytes.toString(userInput);
1009 }
1010
1011
1012
1013
1014
1015
1016
1017
1018 public BigInteger split2(BigInteger a, BigInteger b) {
1019 return a.add(b).divide(BigInteger.valueOf(2)).abs();
1020 }
1021
1022
1023
1024
1025
1026
1027
1028 public byte[][] convertToBytes(BigInteger[] bigIntegers) {
1029 byte[][] returnBytes = new byte[bigIntegers.length][];
1030 for (int i = 0; i < bigIntegers.length; i++) {
1031 returnBytes[i] = convertToByte(bigIntegers[i]);
1032 }
1033 return returnBytes;
1034 }
1035
1036
1037
1038
1039
1040
1041
1042
1043 public static byte[] convertToByte(BigInteger bigInteger, int pad) {
1044 String bigIntegerString = bigInteger.toString(16);
1045 bigIntegerString = StringUtils.leftPad(bigIntegerString, pad, '0');
1046 return Bytes.toBytes(bigIntegerString);
1047 }
1048
1049
1050
1051
1052
1053
1054
1055 public byte[] convertToByte(BigInteger bigInteger) {
1056 return convertToByte(bigInteger, rowComparisonLength);
1057 }
1058
1059
1060
1061
1062
1063
1064
1065 public BigInteger convertToBigInteger(byte[] row) {
1066 return (row.length > 0) ? new BigInteger(Bytes.toString(row), 16)
1067 : BigInteger.ZERO;
1068 }
1069
1070 @Override
1071 public String toString() {
1072 return this.getClass().getSimpleName() + " [" + rowToStr(firstRow())
1073 + "," + rowToStr(lastRow()) + "]";
1074 }
1075 }
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085 public static class UniformSplit implements SplitAlgorithm {
1086 static final byte xFF = (byte) 0xFF;
1087 byte[] firstRowBytes = ArrayUtils.EMPTY_BYTE_ARRAY;
1088 byte[] lastRowBytes =
1089 new byte[] {xFF, xFF, xFF, xFF, xFF, xFF, xFF, xFF};
1090 public byte[] split(byte[] start, byte[] end) {
1091 return Bytes.split(start, end, 1)[1];
1092 }
1093
1094 @Override
1095 public byte[][] split(int numRegions) {
1096 Preconditions.checkArgument(
1097 Bytes.compareTo(lastRowBytes, firstRowBytes) > 0,
1098 "last row (%s) is configured less than first row (%s)",
1099 Bytes.toStringBinary(lastRowBytes),
1100 Bytes.toStringBinary(firstRowBytes));
1101
1102 byte[][] splits = Bytes.split(firstRowBytes, lastRowBytes, true,
1103 numRegions - 1);
1104 Preconditions.checkState(splits != null,
1105 "Could not split region with given user input: " + this);
1106
1107
1108 return Arrays.copyOfRange(splits, 1, splits.length - 1);
1109 }
1110
1111 public byte[][] split(byte[] start, byte[] end, int numSplits, boolean inclusive) {
1112 if (Arrays.equals(start, HConstants.EMPTY_BYTE_ARRAY)) {
1113 start = firstRowBytes;
1114 }
1115 if (Arrays.equals(end, HConstants.EMPTY_BYTE_ARRAY)) {
1116 end = lastRowBytes;
1117 }
1118 Preconditions.checkArgument(
1119 Bytes.compareTo(end, start) > 0,
1120 "last row (%s) is configured less than first row (%s)",
1121 Bytes.toStringBinary(end),
1122 Bytes.toStringBinary(start));
1123
1124 byte[][] splits = Bytes.split(start, end, true,
1125 numSplits - 1);
1126 Preconditions.checkState(splits != null,
1127 "Could not calculate input splits with given user input: " + this);
1128 if (inclusive) {
1129 return splits;
1130 } else {
1131
1132 return Arrays.copyOfRange(splits, 1, splits.length - 1);
1133 }
1134 }
1135
1136 @Override
1137 public byte[] firstRow() {
1138 return firstRowBytes;
1139 }
1140
1141 @Override
1142 public byte[] lastRow() {
1143 return lastRowBytes;
1144 }
1145
1146 @Override
1147 public void setFirstRow(String userInput) {
1148 firstRowBytes = Bytes.toBytesBinary(userInput);
1149 }
1150
1151 @Override
1152 public void setLastRow(String userInput) {
1153 lastRowBytes = Bytes.toBytesBinary(userInput);
1154 }
1155
1156
1157 @Override
1158 public void setFirstRow(byte[] userInput) {
1159 firstRowBytes = userInput;
1160 }
1161
1162 @Override
1163 public void setLastRow(byte[] userInput) {
1164 lastRowBytes = userInput;
1165 }
1166
1167 @Override
1168 public byte[] strToRow(String input) {
1169 return Bytes.toBytesBinary(input);
1170 }
1171
1172 @Override
1173 public String rowToStr(byte[] row) {
1174 return Bytes.toStringBinary(row);
1175 }
1176
1177 @Override
1178 public String separator() {
1179 return ",";
1180 }
1181
1182 @Override
1183 public String toString() {
1184 return this.getClass().getSimpleName() + " [" + rowToStr(firstRow())
1185 + "," + rowToStr(lastRow()) + "]";
1186 }
1187 }
1188 }