1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.mapreduce;
20
21 import static org.junit.Assert.assertEquals;
22
23 import com.google.common.base.Joiner;
24
25 import com.google.common.collect.Sets;
26 import org.apache.commons.cli.CommandLine;
27 import org.apache.commons.lang.RandomStringUtils;
28 import org.apache.commons.logging.Log;
29 import org.apache.commons.logging.LogFactory;
30 import org.apache.hadoop.conf.Configuration;
31 import org.apache.hadoop.fs.Path;
32 import org.apache.hadoop.hbase.Cell;
33 import org.apache.hadoop.hbase.CellUtil;
34 import org.apache.hadoop.hbase.HBaseConfiguration;
35 import org.apache.hadoop.hbase.HBaseTestingUtility;
36 import org.apache.hadoop.hbase.HTableDescriptor;
37 import org.apache.hadoop.hbase.IntegrationTestBase;
38 import org.apache.hadoop.hbase.IntegrationTestingUtility;
39 import org.apache.hadoop.hbase.KeyValue;
40 import org.apache.hadoop.hbase.TableName;
41 import org.apache.hadoop.hbase.client.Admin;
42 import org.apache.hadoop.hbase.client.Connection;
43 import org.apache.hadoop.hbase.client.ConnectionFactory;
44 import org.apache.hadoop.hbase.client.Consistency;
45 import org.apache.hadoop.hbase.client.RegionLocator;
46 import org.apache.hadoop.hbase.client.Result;
47 import org.apache.hadoop.hbase.client.Scan;
48 import org.apache.hadoop.hbase.client.Table;
49 import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
50 import org.apache.hadoop.hbase.coprocessor.ObserverContext;
51 import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
52 import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
53 import org.apache.hadoop.hbase.regionserver.InternalScanner;
54 import org.apache.hadoop.hbase.regionserver.RegionScanner;
55 import org.apache.hadoop.hbase.testclassification.IntegrationTests;
56 import org.apache.hadoop.hbase.util.Bytes;
57 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
58 import org.apache.hadoop.hbase.util.RegionSplitter;
59 import org.apache.hadoop.io.LongWritable;
60 import org.apache.hadoop.io.NullWritable;
61 import org.apache.hadoop.io.Writable;
62 import org.apache.hadoop.io.WritableComparable;
63 import org.apache.hadoop.io.WritableComparator;
64 import org.apache.hadoop.io.WritableUtils;
65 import org.apache.hadoop.mapreduce.InputFormat;
66 import org.apache.hadoop.mapreduce.InputSplit;
67 import org.apache.hadoop.mapreduce.Job;
68 import org.apache.hadoop.mapreduce.JobContext;
69 import org.apache.hadoop.mapreduce.Mapper;
70 import org.apache.hadoop.mapreduce.Partitioner;
71 import org.apache.hadoop.mapreduce.RecordReader;
72 import org.apache.hadoop.mapreduce.Reducer;
73 import org.apache.hadoop.mapreduce.TaskAttemptContext;
74 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
75 import org.apache.hadoop.util.StringUtils;
76 import org.apache.hadoop.util.ToolRunner;
77 import org.junit.Test;
78 import org.junit.experimental.categories.Category;
79
80 import java.io.DataInput;
81 import java.io.DataOutput;
82 import java.io.IOException;
83 import java.util.ArrayList;
84 import java.util.HashSet;
85 import java.util.List;
86 import java.util.Map;
87 import java.util.Random;
88 import java.util.Set;
89 import java.util.concurrent.atomic.AtomicLong;
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 @Category(IntegrationTests.class)
125 public class IntegrationTestBulkLoad extends IntegrationTestBase {
126
127 private static final Log LOG = LogFactory.getLog(IntegrationTestBulkLoad.class);
128
129 private static final byte[] CHAIN_FAM = Bytes.toBytes("L");
130 private static final byte[] SORT_FAM = Bytes.toBytes("S");
131 private static final byte[] DATA_FAM = Bytes.toBytes("D");
132
133 private static String CHAIN_LENGTH_KEY = "hbase.IntegrationTestBulkLoad.chainLength";
134 private static int CHAIN_LENGTH = 500000;
135
136 private static String NUM_MAPS_KEY = "hbase.IntegrationTestBulkLoad.numMaps";
137 private static int NUM_MAPS = 1;
138
139 private static String NUM_IMPORT_ROUNDS_KEY = "hbase.IntegrationTestBulkLoad.numImportRounds";
140 private static int NUM_IMPORT_ROUNDS = 1;
141
142 private static String ROUND_NUM_KEY = "hbase.IntegrationTestBulkLoad.roundNum";
143
144 private static String TABLE_NAME_KEY = "hbase.IntegrationTestBulkLoad.tableName";
145 private static String TABLE_NAME = "IntegrationTestBulkLoad";
146
147 private static String NUM_REPLICA_COUNT_KEY = "hbase.IntegrationTestBulkLoad.replicaCount";
148 private static int NUM_REPLICA_COUNT_DEFAULT = 1;
149
150 private static final String OPT_LOAD = "load";
151 private static final String OPT_CHECK = "check";
152
153 private boolean load = false;
154 private boolean check = false;
155
156 public static class SlowMeCoproScanOperations extends BaseRegionObserver {
157 static final AtomicLong sleepTime = new AtomicLong(2000);
158 Random r = new Random();
159 AtomicLong countOfNext = new AtomicLong(0);
160 AtomicLong countOfOpen = new AtomicLong(0);
161 public SlowMeCoproScanOperations() {}
162 @Override
163 public RegionScanner preScannerOpen(final ObserverContext<RegionCoprocessorEnvironment> e,
164 final Scan scan, final RegionScanner s) throws IOException {
165 if (countOfOpen.incrementAndGet() == 2) {
166 slowdownCode(e);
167 }
168 return s;
169 }
170
171 @Override
172 public boolean preScannerNext(final ObserverContext<RegionCoprocessorEnvironment> e,
173 final InternalScanner s, final List<Result> results,
174 final int limit, final boolean hasMore) throws IOException {
175
176
177 countOfNext.incrementAndGet();
178 if (countOfNext.get() == 0 || countOfNext.get() == 4) {
179 slowdownCode(e);
180 }
181 return true;
182 }
183 protected void slowdownCode(final ObserverContext<RegionCoprocessorEnvironment> e) {
184 if (e.getEnvironment().getRegion().getRegionInfo().getReplicaId() == 0) {
185 try {
186 if (sleepTime.get() > 0) {
187 LOG.info("Sleeping for " + sleepTime.get() + " ms");
188 Thread.sleep(sleepTime.get());
189 }
190 } catch (InterruptedException e1) {
191 LOG.error(e1);
192 }
193 }
194 }
195 }
196
197
198
199
200 private void installSlowingCoproc() throws IOException, InterruptedException {
201 int replicaCount = conf.getInt(NUM_REPLICA_COUNT_KEY, NUM_REPLICA_COUNT_DEFAULT);
202 if (replicaCount == NUM_REPLICA_COUNT_DEFAULT) return;
203
204 TableName t = getTablename();
205 Admin admin = util.getHBaseAdmin();
206 HTableDescriptor desc = admin.getTableDescriptor(t);
207 desc.addCoprocessor(SlowMeCoproScanOperations.class.getName());
208 HBaseTestingUtility.modifyTableSync(admin, desc);
209
210
211 Thread.sleep(30000);
212 }
213
214 @Test
215 public void testBulkLoad() throws Exception {
216 runLoad();
217 installSlowingCoproc();
218 runCheckWithRetry();
219 }
220
221 public void runLoad() throws Exception {
222 setupTable();
223 int numImportRounds = getConf().getInt(NUM_IMPORT_ROUNDS_KEY, NUM_IMPORT_ROUNDS);
224 LOG.info("Running load with numIterations:" + numImportRounds);
225 for (int i = 0; i < numImportRounds; i++) {
226 runLinkedListMRJob(i);
227 }
228 }
229
230 private byte[][] getSplits(int numRegions) {
231 RegionSplitter.UniformSplit split = new RegionSplitter.UniformSplit();
232 split.setFirstRow(Bytes.toBytes(0L));
233 split.setLastRow(Bytes.toBytes(Long.MAX_VALUE));
234 return split.split(numRegions);
235 }
236
237 private void setupTable() throws IOException, InterruptedException {
238 if (util.getHBaseAdmin().tableExists(getTablename())) {
239 util.deleteTable(getTablename());
240 }
241
242 util.createTable(
243 getTablename().getName(),
244 new byte[][]{CHAIN_FAM, SORT_FAM, DATA_FAM},
245 getSplits(16)
246 );
247
248 int replicaCount = conf.getInt(NUM_REPLICA_COUNT_KEY, NUM_REPLICA_COUNT_DEFAULT);
249 if (replicaCount == NUM_REPLICA_COUNT_DEFAULT) return;
250
251 TableName t = getTablename();
252 HBaseTestingUtility.setReplicas(util.getHBaseAdmin(), t, replicaCount);
253 }
254
255 private void runLinkedListMRJob(int iteration) throws Exception {
256 String jobName = IntegrationTestBulkLoad.class.getSimpleName() + " - " +
257 EnvironmentEdgeManager.currentTime();
258 Configuration conf = new Configuration(util.getConfiguration());
259 Path p = null;
260 if (conf.get(ImportTsv.BULK_OUTPUT_CONF_KEY) == null) {
261 p = util.getDataTestDirOnTestFS(getTablename() + "-" + iteration);
262 } else {
263 p = new Path(conf.get(ImportTsv.BULK_OUTPUT_CONF_KEY));
264 }
265
266 conf.setBoolean("mapreduce.map.speculative", false);
267 conf.setBoolean("mapreduce.reduce.speculative", false);
268 conf.setInt(ROUND_NUM_KEY, iteration);
269
270 Job job = new Job(conf);
271
272 job.setJobName(jobName);
273
274
275 job.setInputFormatClass(ITBulkLoadInputFormat.class);
276
277
278 job.setMapperClass(LinkedListCreationMapper.class);
279 job.setMapOutputKeyClass(ImmutableBytesWritable.class);
280 job.setMapOutputValueClass(KeyValue.class);
281
282
283
284
285
286 job.setJarByClass(getClass());
287
288
289 FileOutputFormat.setOutputPath(job, p);
290 try (Connection conn = ConnectionFactory.createConnection(conf);
291 Admin admin = conn.getAdmin();
292 Table table = conn.getTable(getTablename());
293 RegionLocator regionLocator = conn.getRegionLocator(getTablename())) {
294
295
296 HFileOutputFormat2.configureIncrementalLoad(job, table.getTableDescriptor(), regionLocator);
297
298
299 assertEquals(true, job.waitForCompletion(true));
300
301
302 LoadIncrementalHFiles loader = new LoadIncrementalHFiles(conf);
303
304
305 loader.doBulkLoad(p, admin, table, regionLocator);
306 }
307
308
309 util.getTestFileSystem().delete(p, true);
310 }
311
312 public static class EmptySplit extends InputSplit implements Writable {
313 @Override
314 public void write(DataOutput out) throws IOException { }
315 @Override
316 public void readFields(DataInput in) throws IOException { }
317 @Override
318 public long getLength() { return 0L; }
319 @Override
320 public String[] getLocations() { return new String[0]; }
321 }
322
323 public static class FixedRecordReader<K, V> extends RecordReader<K, V> {
324 private int index = -1;
325 private K[] keys;
326 private V[] values;
327
328 public FixedRecordReader(K[] keys, V[] values) {
329 this.keys = keys;
330 this.values = values;
331 }
332 @Override
333 public void initialize(InputSplit split, TaskAttemptContext context) throws IOException,
334 InterruptedException { }
335 @Override
336 public boolean nextKeyValue() throws IOException, InterruptedException {
337 return ++index < keys.length;
338 }
339 @Override
340 public K getCurrentKey() throws IOException, InterruptedException {
341 return keys[index];
342 }
343 @Override
344 public V getCurrentValue() throws IOException, InterruptedException {
345 return values[index];
346 }
347 @Override
348 public float getProgress() throws IOException, InterruptedException {
349 return (float)index / keys.length;
350 }
351 @Override
352 public void close() throws IOException {
353 }
354 }
355
356 public static class ITBulkLoadInputFormat extends InputFormat<LongWritable, LongWritable> {
357 @Override
358 public List<InputSplit> getSplits(JobContext context) throws IOException, InterruptedException {
359 int numSplits = context.getConfiguration().getInt(NUM_MAPS_KEY, NUM_MAPS);
360 ArrayList<InputSplit> ret = new ArrayList<InputSplit>(numSplits);
361 for (int i = 0; i < numSplits; ++i) {
362 ret.add(new EmptySplit());
363 }
364 return ret;
365 }
366
367 @Override
368 public RecordReader<LongWritable, LongWritable> createRecordReader(InputSplit split,
369 TaskAttemptContext context)
370 throws IOException, InterruptedException {
371 int taskId = context.getTaskAttemptID().getTaskID().getId();
372 int numMapTasks = context.getConfiguration().getInt(NUM_MAPS_KEY, NUM_MAPS);
373 int numIterations = context.getConfiguration().getInt(NUM_IMPORT_ROUNDS_KEY, NUM_IMPORT_ROUNDS);
374 int iteration = context.getConfiguration().getInt(ROUND_NUM_KEY, 0);
375
376 taskId = taskId + iteration * numMapTasks;
377 numMapTasks = numMapTasks * numIterations;
378
379 long chainId = Math.abs(new Random().nextLong());
380 chainId = chainId - (chainId % numMapTasks) + taskId;
381 LongWritable[] keys = new LongWritable[] {new LongWritable(chainId)};
382
383 return new FixedRecordReader<LongWritable, LongWritable>(keys, keys);
384 }
385 }
386
387
388
389
390
391
392
393
394 public static class LinkedListCreationMapper
395 extends Mapper<LongWritable, LongWritable, ImmutableBytesWritable, KeyValue> {
396
397 private Random rand = new Random();
398
399 @Override
400 protected void map(LongWritable key, LongWritable value, Context context)
401 throws IOException, InterruptedException {
402 long chainId = value.get();
403 LOG.info("Starting mapper with chainId:" + chainId);
404
405 byte[] chainIdArray = Bytes.toBytes(chainId);
406 long currentRow = 0;
407
408 long chainLength = context.getConfiguration().getLong(CHAIN_LENGTH_KEY, CHAIN_LENGTH);
409 long nextRow = getNextRow(0, chainLength);
410
411 for (long i = 0; i < chainLength; i++) {
412 byte[] rk = Bytes.toBytes(currentRow);
413
414
415 KeyValue linkKv = new KeyValue(rk, CHAIN_FAM, chainIdArray, Bytes.toBytes(nextRow));
416
417 KeyValue sortKv = new KeyValue(rk, SORT_FAM, chainIdArray, Bytes.toBytes(i));
418
419 KeyValue dataKv = new KeyValue(rk, DATA_FAM, chainIdArray,
420 Bytes.toBytes(RandomStringUtils.randomAlphabetic(50))
421 );
422
423
424 context.write(new ImmutableBytesWritable(rk), linkKv);
425 context.write(new ImmutableBytesWritable(rk), sortKv);
426 context.write(new ImmutableBytesWritable(rk), dataKv);
427
428 currentRow = nextRow;
429 nextRow = getNextRow(i+1, chainLength);
430 }
431 }
432
433
434 private long getNextRow(long index, long chainLength) {
435 long nextRow = Math.abs(rand.nextLong());
436
437
438
439 nextRow = nextRow - (nextRow % chainLength) + index;
440 return nextRow;
441 }
442 }
443
444
445
446
447
448
449 public static class LinkKey implements WritableComparable<LinkKey> {
450
451 private Long chainId;
452
453 public Long getOrder() {
454 return order;
455 }
456
457 public Long getChainId() {
458 return chainId;
459 }
460
461 private Long order;
462
463 public LinkKey() {}
464
465 public LinkKey(long chainId, long order) {
466 this.chainId = chainId;
467 this.order = order;
468 }
469
470 @Override
471 public int compareTo(LinkKey linkKey) {
472 int res = getChainId().compareTo(linkKey.getChainId());
473 if (res == 0) {
474 res = getOrder().compareTo(linkKey.getOrder());
475 }
476 return res;
477 }
478
479 @Override
480 public void write(DataOutput dataOutput) throws IOException {
481 WritableUtils.writeVLong(dataOutput, chainId);
482 WritableUtils.writeVLong(dataOutput, order);
483 }
484
485 @Override
486 public void readFields(DataInput dataInput) throws IOException {
487 chainId = WritableUtils.readVLong(dataInput);
488 order = WritableUtils.readVLong(dataInput);
489 }
490 }
491
492
493
494
495 public static class LinkChain implements WritableComparable<LinkChain> {
496
497 public Long getNext() {
498 return next;
499 }
500
501 public Long getRk() {
502 return rk;
503 }
504
505 public LinkChain() {}
506
507 public LinkChain(Long rk, Long next) {
508 this.rk = rk;
509 this.next = next;
510 }
511
512 private Long rk;
513 private Long next;
514
515 @Override
516 public int compareTo(LinkChain linkChain) {
517 int res = getRk().compareTo(linkChain.getRk());
518 if (res == 0) {
519 res = getNext().compareTo(linkChain.getNext());
520 }
521 return res;
522 }
523
524 @Override
525 public void write(DataOutput dataOutput) throws IOException {
526 WritableUtils.writeVLong(dataOutput, rk);
527 WritableUtils.writeVLong(dataOutput, next);
528 }
529
530 @Override
531 public void readFields(DataInput dataInput) throws IOException {
532 rk = WritableUtils.readVLong(dataInput);
533 next = WritableUtils.readVLong(dataInput);
534 }
535 }
536
537
538
539
540
541 public static class NaturalKeyPartitioner extends Partitioner<LinkKey, LinkChain> {
542 @Override
543 public int getPartition(LinkKey linkKey,
544 LinkChain linkChain,
545 int numPartitions) {
546 int hash = linkKey.getChainId().hashCode();
547 return Math.abs(hash % numPartitions);
548 }
549 }
550
551
552
553
554
555 public static class NaturalKeyGroupingComparator extends WritableComparator {
556
557 protected NaturalKeyGroupingComparator() {
558 super(LinkKey.class, true);
559 }
560
561 @Override
562 public int compare(WritableComparable w1, WritableComparable w2) {
563 LinkKey k1 = (LinkKey) w1;
564 LinkKey k2 = (LinkKey) w2;
565
566 return k1.getChainId().compareTo(k2.getChainId());
567 }
568 }
569
570
571
572
573
574 public static class CompositeKeyComparator extends WritableComparator {
575
576 protected CompositeKeyComparator() {
577 super(LinkKey.class, true);
578 }
579
580 @Override
581 public int compare(WritableComparable w1, WritableComparable w2) {
582 LinkKey k1 = (LinkKey) w1;
583 LinkKey k2 = (LinkKey) w2;
584
585 return k1.compareTo(k2);
586 }
587 }
588
589
590
591
592
593
594
595 public static class LinkedListCheckingMapper extends TableMapper<LinkKey, LinkChain> {
596 @Override
597 protected void map(ImmutableBytesWritable key, Result value, Context context)
598 throws IOException, InterruptedException {
599 long longRk = Bytes.toLong(value.getRow());
600
601 for (Map.Entry<byte[], byte[]> entry : value.getFamilyMap(CHAIN_FAM).entrySet()) {
602 long chainId = Bytes.toLong(entry.getKey());
603 long next = Bytes.toLong(entry.getValue());
604 Cell c = value.getColumnCells(SORT_FAM, entry.getKey()).get(0);
605 long order = Bytes.toLong(CellUtil.cloneValue(c));
606 context.write(new LinkKey(chainId, order), new LinkChain(longRk, next));
607 }
608 }
609 }
610
611
612
613
614
615
616
617
618
619
620 public static class LinkedListCheckingReducer
621 extends Reducer<LinkKey, LinkChain, NullWritable, NullWritable> {
622 @Override
623 protected void reduce(LinkKey key, Iterable<LinkChain> values, Context context)
624 throws java.io.IOException, java.lang.InterruptedException {
625 long next = -1L;
626 long prev = -1L;
627 long count = 0L;
628
629 for (LinkChain lc : values) {
630
631 if (next == -1) {
632 if (lc.getRk() != 0L) {
633 String msg = "Chains should all start at rk 0, but read rk " + lc.getRk()
634 + ". Chain:" + key.chainId + ", order:" + key.order;
635 logError(msg, context);
636 throw new RuntimeException(msg);
637 }
638 next = lc.getNext();
639 } else {
640 if (next != lc.getRk()) {
641 String msg = "Missing a link in the chain. Prev rk " + prev + " was, expecting "
642 + next + " but got " + lc.getRk() + ". Chain:" + key.chainId
643 + ", order:" + key.order;
644 logError(msg, context);
645 throw new RuntimeException(msg);
646 }
647 prev = lc.getRk();
648 next = lc.getNext();
649 }
650 count++;
651 }
652
653 int expectedChainLen = context.getConfiguration().getInt(CHAIN_LENGTH_KEY, CHAIN_LENGTH);
654 if (count != expectedChainLen) {
655 String msg = "Chain wasn't the correct length. Expected " + expectedChainLen + " got "
656 + count + ". Chain:" + key.chainId + ", order:" + key.order;
657 logError(msg, context);
658 throw new RuntimeException(msg);
659 }
660 }
661
662 private static void logError(String msg, Context context) throws IOException {
663 TableName table = getTableName(context.getConfiguration());
664
665 LOG.error("Failure in chain verification: " + msg);
666 try (Connection connection = ConnectionFactory.createConnection(context.getConfiguration());
667 Admin admin = connection.getAdmin()) {
668 LOG.error("cluster status:\n" + admin.getClusterStatus());
669 LOG.error("table regions:\n"
670 + Joiner.on("\n").join(admin.getTableRegions(table)));
671 }
672 }
673 }
674
675 private void runCheckWithRetry() throws IOException, ClassNotFoundException, InterruptedException {
676 try {
677 runCheck();
678 } catch (Throwable t) {
679 LOG.warn("Received " + StringUtils.stringifyException(t));
680 LOG.warn("Running the check MR Job again to see whether an ephemeral problem or not");
681 runCheck();
682 throw t;
683 }
684
685 }
686
687
688
689
690
691
692
693
694 private void runCheck() throws IOException, ClassNotFoundException, InterruptedException {
695 LOG.info("Running check");
696 Configuration conf = getConf();
697 String jobName = getTablename() + "_check" + EnvironmentEdgeManager.currentTime();
698 Path p = util.getDataTestDirOnTestFS(jobName);
699
700 Job job = new Job(conf);
701 job.setJarByClass(getClass());
702 job.setJobName(jobName);
703
704 job.setPartitionerClass(NaturalKeyPartitioner.class);
705 job.setGroupingComparatorClass(NaturalKeyGroupingComparator.class);
706 job.setSortComparatorClass(CompositeKeyComparator.class);
707
708 Scan scan = new Scan();
709 scan.addFamily(CHAIN_FAM);
710 scan.addFamily(SORT_FAM);
711 scan.setMaxVersions(1);
712 scan.setCacheBlocks(false);
713 scan.setBatch(1000);
714
715 int replicaCount = conf.getInt(NUM_REPLICA_COUNT_KEY, NUM_REPLICA_COUNT_DEFAULT);
716 if (replicaCount != NUM_REPLICA_COUNT_DEFAULT) {
717 scan.setConsistency(Consistency.TIMELINE);
718 }
719
720 TableMapReduceUtil.initTableMapperJob(
721 getTablename().getName(),
722 scan,
723 LinkedListCheckingMapper.class,
724 LinkKey.class,
725 LinkChain.class,
726 job
727 );
728
729 job.setReducerClass(LinkedListCheckingReducer.class);
730 job.setOutputKeyClass(NullWritable.class);
731 job.setOutputValueClass(NullWritable.class);
732
733 FileOutputFormat.setOutputPath(job, p);
734
735 assertEquals(true, job.waitForCompletion(true));
736
737
738 util.getTestFileSystem().delete(p, true);
739 }
740
741 @Override
742 public void setUpCluster() throws Exception {
743 util = getTestingUtil(getConf());
744 util.initializeCluster(1);
745 int replicaCount = getConf().getInt(NUM_REPLICA_COUNT_KEY, NUM_REPLICA_COUNT_DEFAULT);
746 if (LOG.isDebugEnabled() && replicaCount != NUM_REPLICA_COUNT_DEFAULT) {
747 LOG.debug("Region Replicas enabled: " + replicaCount);
748 }
749
750
751 if (util.isDistributedCluster()) {
752 util.getConfiguration().setIfUnset(NUM_MAPS_KEY,
753 Integer.toString(util.getHBaseAdmin().getClusterStatus().getServersSize() * 10)
754 );
755 util.getConfiguration().setIfUnset(NUM_IMPORT_ROUNDS_KEY, "5");
756 } else {
757 util.startMiniMapReduceCluster();
758 }
759 }
760
761 @Override
762 protected void addOptions() {
763 super.addOptions();
764 super.addOptNoArg(OPT_CHECK, "Run check only");
765 super.addOptNoArg(OPT_LOAD, "Run load only");
766 }
767
768 @Override
769 protected void processOptions(CommandLine cmd) {
770 super.processOptions(cmd);
771 check = cmd.hasOption(OPT_CHECK);
772 load = cmd.hasOption(OPT_LOAD);
773 }
774
775 @Override
776 public int runTestFromCommandLine() throws Exception {
777 if (load) {
778 runLoad();
779 } else if (check) {
780 installSlowingCoproc();
781 runCheckWithRetry();
782 } else {
783 testBulkLoad();
784 }
785 return 0;
786 }
787
788 @Override
789 public TableName getTablename() {
790 return getTableName(getConf());
791 }
792
793 public static TableName getTableName(Configuration conf) {
794 return TableName.valueOf(conf.get(TABLE_NAME_KEY, TABLE_NAME));
795 }
796
797 @Override
798 protected Set<String> getColumnFamilies() {
799 return Sets.newHashSet(Bytes.toString(CHAIN_FAM) , Bytes.toString(DATA_FAM),
800 Bytes.toString(SORT_FAM));
801 }
802
803 public static void main(String[] args) throws Exception {
804 Configuration conf = HBaseConfiguration.create();
805 IntegrationTestingUtility.setUseDistributedCluster(conf);
806 int status = ToolRunner.run(conf, new IntegrationTestBulkLoad(), args);
807 System.exit(status);
808 }
809 }