View Javadoc

1   /**
2    *
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *     http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing, software
14   * distributed under the License is distributed on an "AS IS" BASIS,
15   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16   * See the License for the specific language governing permissions and
17   * limitations under the License.
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   * The {@link RegionSplitter} class provides several utilities to help in the
74   * administration lifecycle for developers who choose to manually split regions
75   * instead of having HBase handle that automatically. The most useful utilities
76   * are:
77   * <p>
78   * <ul>
79   * <li>Create a table with a specified number of pre-split regions
80   * <li>Execute a rolling split of all regions on an existing table
81   * </ul>
82   * <p>
83   * Both operations can be safely done on a live server.
84   * <p>
85   * <b>Question:</b> How do I turn off automatic splitting? <br>
86   * <b>Answer:</b> Automatic splitting is determined by the configuration value
87   * <i>HConstants.HREGION_MAX_FILESIZE</i>. It is not recommended that you set this
88   * to Long.MAX_VALUE in case you forget about manual splits. A suggested setting
89   * is 100GB, which would result in &gt; 1hr major compactions if reached.
90   * <p>
91   * <b>Question:</b> Why did the original authors decide to manually split? <br>
92   * <b>Answer:</b> Specific workload characteristics of our use case allowed us
93   * to benefit from a manual split system.
94   * <p>
95   * <ul>
96   * <li>Data (~1k) that would grow instead of being replaced
97   * <li>Data growth was roughly uniform across all regions
98   * <li>OLTP workload. Data loss is a big deal.
99   * </ul>
100  * <p>
101  * <b>Question:</b> Why is manual splitting good for this workload? <br>
102  * <b>Answer:</b> Although automated splitting is not a bad option, there are
103  * benefits to manual splitting.
104  * <p>
105  * <ul>
106  * <li>With growing amounts of data, splits will continually be needed. Since
107  * you always know exactly what regions you have, long-term debugging and
108  * profiling is much easier with manual splits. It is hard to trace the logs to
109  * understand region level problems if it keeps splitting and getting renamed.
110  * <li>Data offlining bugs + unknown number of split regions == oh crap! If an
111  * WAL or StoreFile was mistakenly unprocessed by HBase due to a weird bug and
112  * you notice it a day or so later, you can be assured that the regions
113  * specified in these files are the same as the current regions and you have
114  * less headaches trying to restore/replay your data.
115  * <li>You can finely tune your compaction algorithm. With roughly uniform data
116  * growth, it's easy to cause split / compaction storms as the regions all
117  * roughly hit the same data size at the same time. With manual splits, you can
118  * let staggered, time-based major compactions spread out your network IO load.
119  * </ul>
120  * <p>
121  * <b>Question:</b> What's the optimal number of pre-split regions to create? <br>
122  * <b>Answer:</b> Mileage will vary depending upon your application.
123  * <p>
124  * The short answer for our application is that we started with 10 pre-split
125  * regions / server and watched our data growth over time. It's better to err on
126  * the side of too little regions and rolling split later.
127  * <p>
128  * The more complicated answer is that this depends upon the largest storefile
129  * in your region. With a growing data size, this will get larger over time. You
130  * want the largest region to be just big enough that the
131  * {@link org.apache.hadoop.hbase.regionserver.HStore} compact
132  * selection algorithm only compacts it due to a timed major. If you don't, your
133  * cluster can be prone to compaction storms as the algorithm decides to run
134  * major compactions on a large series of regions all at once. Note that
135  * compaction storms are due to the uniform data growth, not the manual split
136  * decision.
137  * <p>
138  * If you pre-split your regions too thin, you can increase the major compaction
139  * interval by configuring HConstants.MAJOR_COMPACTION_PERIOD. If your data size
140  * grows too large, use this script to perform a network IO safe rolling split
141  * of all regions.
142  */
143 @InterfaceAudience.Private
144 public class RegionSplitter {
145   private static final Log LOG = LogFactory.getLog(RegionSplitter.class);
146 
147   /**
148    * A generic interface for the RegionSplitter code to use for all it's
149    * functionality. Note that the original authors of this code use
150    * {@link HexStringSplit} to partition their table and set it as default, but
151    * provided this for your custom algorithm. To use, create a new derived class
152    * from this interface and call {@link RegionSplitter#createPresplitTable} or
153    * RegionSplitter#rollingSplit(TableName, SplitAlgorithm, Configuration) with the
154    * argument splitClassName giving the name of your class.
155    */
156   public interface SplitAlgorithm {
157     /**
158      * Split a pre-existing region into 2 regions.
159      *
160      * @param start
161      *          first row (inclusive)
162      * @param end
163      *          last row (exclusive)
164      * @return the split row to use
165      */
166     byte[] split(byte[] start, byte[] end);
167 
168     /**
169      * Split an entire table.
170      *
171      * @param numRegions
172      *          number of regions to split the table into
173      *
174      * @throws RuntimeException
175      *           user input is validated at this time. may throw a runtime
176      *           exception in response to a parse failure
177      * @return array of split keys for the initial regions of the table. The
178      *         length of the returned array should be numRegions-1.
179      */
180     byte[][] split(int numRegions);
181 
182     /**
183      * Some MapReduce jobs may want to run multiple mappers per region,
184      * this is intended for such usecase.
185      *
186      * @param start first row (inclusive)
187      * @param end last row (exclusive)
188      * @param numSplits number of splits to generate
189      * @param inclusive whether start and end are returned as split points
190      */
191     byte[][] split(byte[] start, byte[] end, int numSplits, boolean inclusive);
192 
193     /**
194      * In HBase, the first row is represented by an empty byte array. This might
195      * cause problems with your split algorithm or row printing. All your APIs
196      * will be passed firstRow() instead of empty array.
197      *
198      * @return your representation of your first row
199      */
200     byte[] firstRow();
201 
202     /**
203      * In HBase, the last row is represented by an empty byte array. This might
204      * cause problems with your split algorithm or row printing. All your APIs
205      * will be passed firstRow() instead of empty array.
206      *
207      * @return your representation of your last row
208      */
209     byte[] lastRow();
210 
211     /**
212      * In HBase, the last row is represented by an empty byte array. Set this
213      * value to help the split code understand how to evenly divide the first
214      * region.
215      *
216      * @param userInput
217      *          raw user input (may throw RuntimeException on parse failure)
218      */
219     void setFirstRow(String userInput);
220 
221     /**
222      * In HBase, the last row is represented by an empty byte array. Set this
223      * value to help the split code understand how to evenly divide the last
224      * region. Note that this last row is inclusive for all rows sharing the
225      * same prefix.
226      *
227      * @param userInput
228      *          raw user input (may throw RuntimeException on parse failure)
229      */
230     void setLastRow(String userInput);
231 
232     /**
233      * @param input
234      *          user or file input for row
235      * @return byte array representation of this row for HBase
236      */
237     byte[] strToRow(String input);
238 
239     /**
240      * @param row
241      *          byte array representing a row in HBase
242      * @return String to use for debug &amp; file printing
243      */
244     String rowToStr(byte[] row);
245 
246     /**
247      * @return the separator character to use when storing / printing the row
248      */
249     String separator();
250 
251     /**
252      * Set the first row
253      * @param userInput byte array of the row key.
254      */
255     void setFirstRow(byte[] userInput);
256 
257     /**
258      * Set the last row
259      * @param userInput byte array of the row key.
260      */
261     void setLastRow(byte[] userInput);
262   }
263 
264   /**
265    * The main function for the RegionSplitter application. Common uses:
266    * <p>
267    * <ul>
268    * <li>create a table named 'myTable' with 60 pre-split regions containing 2
269    * column families 'test' &amp; 'rs', assuming the keys are hex-encoded ASCII:
270    * <ul>
271    * <li>bin/hbase org.apache.hadoop.hbase.util.RegionSplitter -c 60 -f test:rs
272    * myTable HexStringSplit
273    * </ul>
274    * <li>perform a rolling split of 'myTable' (i.e. 60 =&gt; 120 regions), # 2
275    * outstanding splits at a time, assuming keys are uniformly distributed
276    * bytes:
277    * <ul>
278    * <li>bin/hbase org.apache.hadoop.hbase.util.RegionSplitter -r -o 2 myTable
279    * UniformSplit
280    * </ul>
281    * </ul>
282    *
283    * There are two SplitAlgorithms built into RegionSplitter, HexStringSplit
284    * and UniformSplit. These are different strategies for choosing region
285    * boundaries. See their source code for details.
286    *
287    * @param args
288    *          Usage: RegionSplitter &lt;TABLE&gt; &lt;SPLITALGORITHM&gt;
289    *          &lt;-c &lt;# regions&gt; -f &lt;family:family:...&gt; | -r
290    *          [-o &lt;# outstanding splits&gt;]&gt;
291    *          [-D &lt;conf.param=value&gt;]
292    * @throws IOException
293    *           HBase IO problem
294    * @throws InterruptedException
295    *           user requested exit
296    * @throws ParseException
297    *           problem parsing user input
298    */
299   @SuppressWarnings("static-access")
300   public static void main(String[] args) throws IOException,
301       InterruptedException, ParseException {
302     Configuration conf = HBaseConfiguration.create();
303 
304     // parse user input
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         // NOTE: createTable is synchronous on the table, but not on the regions
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); // sleep
416           }
417         }
418       }
419       LOG.debug("Finished creating table with " + splitCount + " regions");
420     }
421   }
422 
423   /**
424    * Alternative getCurrentNrHRS which is no longer available.
425    * @param connection
426    * @return Rough count of regionservers out on cluster.
427    * @throws IOException 
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       // Max outstanding splits. default == 50% of servers
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       // Get a list of daughter regions to create
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       // all splits must compact & we have 1 compact thread, so 2 split
470       // requests to the same RS can stall the outstanding split queue.
471       // To fix, group the regions into an RS pool and round-robin through it
472       LOG.debug("Bucketing regions by regionserver...");
473       TreeMap<String, LinkedList<Pair<byte[], byte[]>>> daughterRegions =
474           Maps.newTreeMap();
475       // Get a regionLocator.  Need it in below.
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         // Open the split file and modify it as splits finish
489         byte[] rawData = readFile(fs, splitFile);
490 
491         FSDataOutputStream splitOut = fs.create(splitFile);
492         try {
493           splitOut.write(rawData);
494 
495           try {
496             // *** split code ***
497             while (!daughterRegions.isEmpty()) {
498               LOG.debug(daughterRegions.size() + " RS have regions to splt.");
499 
500               // Get ServerName to region count mapping
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               // Sort the ServerNames by the number of regions they have
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               // Round-robin through the ServerName list. Choose the lightest-loaded servers
521               // first to keep the master from load-balancing regions as we split.
522               for (String rsLoc : serversLeft) {
523                 Pair<byte[], byte[]> dr = null;
524 
525                 // Find a region in the ServerName list that hasn't been moved
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                   // get current region info
532                   byte[] split = dr.getSecond();
533                   HRegionLocation regionLoc = regionLocator.getRegionLocation(split);
534 
535                   // if this region moved locations
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                     // relocate it, don't use it right now
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                   // make sure this region wasn't already split
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                   // passed all checks! found a good region
566                   break;
567                 }
568                 if (regionList.isEmpty()) {
569                   daughterRegions.remove(rsLoc);
570                 }
571                 if (dr == null)
572                   continue;
573 
574                 // we have a good region, time to split!
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                   // we need to verify and rate-limit our splits
585                   outstanding.addLast(dr);
586                   // with too many outstanding splits, wait for some to finish
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                 // mark each finished region as successfully split.
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    * @throws IOException if the specified SplitAlgorithm class couldn't be
656    * instantiated
657    */
658   public static SplitAlgorithm newSplitAlgoInstance(Configuration conf,
659           String splitClassName) throws IOException {
660     Class<?> splitClass;
661 
662     // For split algorithms builtin to RegionSplitter, the user can specify
663     // their simple class name instead of a fully qualified class name.
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     // Get table info
700     Pair<Path, Path> tableDirAndSplitFile =
701       getTableDirAndSplitFile(connection.getConfiguration(), tableName);
702     Path tableDir = tableDirAndSplitFile.getFirst();
703     FileSystem fs = tableDir.getFileSystem(connection.getConfiguration());
704     // Clear the cache to forcibly refresh region information
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       // for every region that hasn't been verified as a finished split
713       for (Pair<byte[], byte[]> region : regionList) {
714         byte[] start = region.getFirst();
715         byte[] split = region.getSecond();
716 
717         // see if the new split daughter region has come online
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           // NSFRE will occur if the old hbase:meta entry has no server assigned
726           LOG.info(nsfre);
727           logicalSplitting.add(region);
728           continue;
729         }
730 
731         try {
732           // when a daughter region is opened, a compaction is triggered
733           // wait until compaction completes for both daughter regions
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             // Check every Column Family for that region -- check does not have references.
746             boolean refFound = false;
747             for (HColumnDescriptor c : htd.getFamilies()) {
748               if ((refFound = regionFs.hasReferences(c.getNameAsString()))) {
749                 break;
750               }
751             }
752 
753             // compaction is completed when all reference files are gone
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    * @param conf
780    * @param tableName
781    * @return A Pair where first item is table dir and second is the split file.
782    * @throws IOException 
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     // Using strings because (new byte[]{0}).equals(new byte[]{0}) == false
804     Set<Pair<String, String>> daughterRegions = Sets.newHashSet();
805 
806     // Does a split file exist?
807     if (!fs.exists(splitFile)) {
808       // NO = fresh start. calculate splits to make
809       LOG.debug("No " + splitFile.getName() + " file. Calculating splits ");
810 
811       // Query meta for all regions in the table
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       // prepare the split file
830       Path tmpFile = new Path(tableDir, "_balancedSplit_prepare");
831       FSDataOutputStream tmpOut = fs.create(tmpFile);
832 
833       // calculate all the splits == [daughterRegions] = [(start, splitPoint)]
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       // parse split file and process remaining splits
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    * HexStringSplit is a well-known {@link SplitAlgorithm} for choosing region
890    * boundaries. The format of a HexStringSplit region boundary is the ASCII
891    * representation of an MD5 checksum, or any other uniformly distributed
892    * hexadecimal value. Row are hex-encoded long values in the range
893    * <b>"00000000" =&gt; "FFFFFFFF"</b> and are left-padded with zeros to keep the
894    * same order lexicographically as if they were binary.
895    *
896    * Since this split algorithm uses hex strings as keys, it is easy to read &amp;
897    * write in the shell but takes up more space and may be non-intuitive.
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       // +1 to range because the last row is inclusive
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         // NOTE: this means the last region gets all the slop.
929         // This is not a big deal if we're assuming n << MAXHEX
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       // +1 to range because the last row is inclusive
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         // NOTE: this means the last region gets all the slop.
953         // This is not a big deal if we're assuming n << MAXHEX
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       // Precondition: lastRow > firstRow, so last's length is the greater
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      * Divide 2 numbers in half (for split algorithm)
1013      *
1014      * @param a number #1
1015      * @param b number #2
1016      * @return the midpoint of the 2 numbers
1017      */
1018     public BigInteger split2(BigInteger a, BigInteger b) {
1019       return a.add(b).divide(BigInteger.valueOf(2)).abs();
1020     }
1021 
1022     /**
1023      * Returns an array of bytes corresponding to an array of BigIntegers
1024      *
1025      * @param bigIntegers numbers to convert
1026      * @return bytes corresponding to the bigIntegers
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      * Returns the bytes corresponding to the BigInteger
1038      *
1039      * @param bigInteger number to convert
1040      * @param pad padding length
1041      * @return byte corresponding to input BigInteger
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      * Returns the bytes corresponding to the BigInteger
1051      *
1052      * @param bigInteger number to convert
1053      * @return corresponding bytes
1054      */
1055     public byte[] convertToByte(BigInteger bigInteger) {
1056       return convertToByte(bigInteger, rowComparisonLength);
1057     }
1058 
1059     /**
1060      * Returns the BigInteger represented by the byte array
1061      *
1062      * @param row byte array representing row
1063      * @return the corresponding BigInteger
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    * A SplitAlgorithm that divides the space of possible keys evenly. Useful
1079    * when the keys are approximately uniform random bytes (e.g. hashes). Rows
1080    * are raw byte values in the range <b>00 =&gt; FF</b> and are right-padded with
1081    * zeros to keep the same memcmp() order. This is the natural algorithm to use
1082    * for a byte[] environment and saves space, but is not necessarily the
1083    * easiest for readability.
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       // remove endpoints, which are included in the splits list
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         // remove endpoints, which are included in the splits list
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 }