1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.test;
20
21 import java.io.DataInput;
22 import java.io.DataOutput;
23 import java.io.FileNotFoundException;
24 import java.io.IOException;
25 import java.io.InterruptedIOException;
26 import java.security.SecureRandom;
27 import java.util.ArrayList;
28 import java.util.Arrays;
29 import java.util.Iterator;
30 import java.util.List;
31 import java.util.Random;
32 import java.util.Set;
33 import java.util.SortedSet;
34 import java.util.TreeSet;
35 import java.util.UUID;
36 import java.util.concurrent.ThreadLocalRandom;
37 import java.util.concurrent.atomic.AtomicInteger;
38
39 import org.apache.commons.cli.CommandLine;
40 import org.apache.commons.cli.GnuParser;
41 import org.apache.commons.cli.HelpFormatter;
42 import org.apache.commons.cli.Options;
43 import org.apache.commons.cli.ParseException;
44 import org.apache.commons.logging.Log;
45 import org.apache.commons.logging.LogFactory;
46 import org.apache.hadoop.conf.Configuration;
47 import org.apache.hadoop.conf.Configured;
48 import org.apache.hadoop.fs.FileSystem;
49 import org.apache.hadoop.fs.LocatedFileStatus;
50 import org.apache.hadoop.fs.Path;
51 import org.apache.hadoop.fs.RemoteIterator;
52 import org.apache.hadoop.hbase.Cell;
53 import org.apache.hadoop.hbase.HBaseConfiguration;
54 import org.apache.hadoop.hbase.HBaseTestingUtility;
55 import org.apache.hadoop.hbase.HColumnDescriptor;
56 import org.apache.hadoop.hbase.HConstants;
57 import org.apache.hadoop.hbase.HRegionLocation;
58 import org.apache.hadoop.hbase.HTableDescriptor;
59 import org.apache.hadoop.hbase.client.HTable;
60 import org.apache.hadoop.hbase.client.Table;
61 import org.apache.hadoop.hbase.IntegrationTestBase;
62 import org.apache.hadoop.hbase.IntegrationTestingUtility;
63 import org.apache.hadoop.hbase.testclassification.IntegrationTests;
64 import org.apache.hadoop.hbase.fs.HFileSystem;
65 import org.apache.hadoop.hbase.MasterNotRunningException;
66 import org.apache.hadoop.hbase.TableName;
67 import org.apache.hadoop.hbase.client.Admin;
68 import org.apache.hadoop.hbase.client.BufferedMutator;
69 import org.apache.hadoop.hbase.client.BufferedMutatorParams;
70 import org.apache.hadoop.hbase.client.Connection;
71 import org.apache.hadoop.hbase.client.ConnectionFactory;
72 import org.apache.hadoop.hbase.client.Get;
73 import org.apache.hadoop.hbase.client.Mutation;
74 import org.apache.hadoop.hbase.client.Put;
75 import org.apache.hadoop.hbase.client.RegionLocator;
76 import org.apache.hadoop.hbase.client.Result;
77 import org.apache.hadoop.hbase.client.ResultScanner;
78 import org.apache.hadoop.hbase.client.Scan;
79 import org.apache.hadoop.hbase.client.ScannerCallable;
80 import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
81 import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
82 import org.apache.hadoop.hbase.mapreduce.TableMapper;
83 import org.apache.hadoop.hbase.mapreduce.TableRecordReaderImpl;
84 import org.apache.hadoop.hbase.mapreduce.WALPlayer;
85 import org.apache.hadoop.hbase.regionserver.FlushLargeStoresPolicy;
86 import org.apache.hadoop.hbase.regionserver.FlushPolicyFactory;
87 import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
88 import org.apache.hadoop.hbase.util.AbstractHBaseTool;
89 import org.apache.hadoop.hbase.util.Bytes;
90 import org.apache.hadoop.hbase.util.RegionSplitter;
91 import org.apache.hadoop.hbase.wal.WALKey;
92 import org.apache.hadoop.io.BytesWritable;
93 import org.apache.hadoop.io.NullWritable;
94 import org.apache.hadoop.io.Writable;
95 import org.apache.hadoop.mapreduce.Counter;
96 import org.apache.hadoop.mapreduce.CounterGroup;
97 import org.apache.hadoop.mapreduce.Counters;
98 import org.apache.hadoop.mapreduce.InputFormat;
99 import org.apache.hadoop.mapreduce.InputSplit;
100 import org.apache.hadoop.mapreduce.Job;
101 import org.apache.hadoop.mapreduce.JobContext;
102 import org.apache.hadoop.mapreduce.Mapper;
103 import org.apache.hadoop.mapreduce.RecordReader;
104 import org.apache.hadoop.mapreduce.Reducer;
105 import org.apache.hadoop.mapreduce.TaskAttemptContext;
106 import org.apache.hadoop.mapreduce.TaskAttemptID;
107 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
108 import org.apache.hadoop.mapreduce.lib.input.FileSplit;
109 import org.apache.hadoop.mapreduce.lib.input.SequenceFileAsBinaryInputFormat;
110 import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
111 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
112 import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;
113 import org.apache.hadoop.mapreduce.lib.output.SequenceFileAsBinaryOutputFormat;
114 import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
115 import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
116 import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl;
117 import org.apache.hadoop.util.Tool;
118 import org.apache.hadoop.util.ToolRunner;
119 import org.junit.Test;
120 import org.junit.experimental.categories.Category;
121
122 import com.google.common.collect.Sets;
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193 @Category(IntegrationTests.class)
194 public class IntegrationTestBigLinkedList extends IntegrationTestBase {
195 protected static final byte[] NO_KEY = new byte[1];
196
197 protected static String TABLE_NAME_KEY = "IntegrationTestBigLinkedList.table";
198
199 protected static String DEFAULT_TABLE_NAME = "IntegrationTestBigLinkedList";
200
201 protected static byte[] FAMILY_NAME = Bytes.toBytes("meta");
202 private static byte[] BIG_FAMILY_NAME = Bytes.toBytes("big");
203 private static byte[] TINY_FAMILY_NAME = Bytes.toBytes("tiny");
204
205
206 protected static final byte[] COLUMN_PREV = Bytes.toBytes("prev");
207
208
209 protected static final byte[] COLUMN_CLIENT = Bytes.toBytes("client");
210
211
212 protected static final byte[] COLUMN_COUNT = Bytes.toBytes("count");
213
214
215 private static final String GENERATOR_NUM_ROWS_PER_MAP_KEY
216 = "IntegrationTestBigLinkedList.generator.num_rows";
217
218 private static final String GENERATOR_NUM_MAPPERS_KEY
219 = "IntegrationTestBigLinkedList.generator.map.tasks";
220
221 private static final String GENERATOR_WIDTH_KEY
222 = "IntegrationTestBigLinkedList.generator.width";
223
224 private static final String GENERATOR_WRAP_KEY
225 = "IntegrationTestBigLinkedList.generator.wrap";
226
227 private static final String CONCURRENT_WALKER_KEY
228 = "IntegrationTestBigLinkedList.generator.concurrentwalkers";
229
230 protected int NUM_SLAVES_BASE = 3;
231
232 private static final int MISSING_ROWS_TO_LOG = 10;
233
234 private static final int WIDTH_DEFAULT = 1000000;
235 private static final int WRAP_DEFAULT = 25;
236 private static final int ROWKEY_LENGTH = 16;
237
238 private static final int CONCURRENT_WALKER_DEFAULT = 0;
239
240 protected String toRun;
241 protected String[] otherArgs;
242
243 static class CINode {
244 byte[] key;
245 byte[] prev;
246 String client;
247 long count;
248 }
249
250
251
252
253 static class Generator extends Configured implements Tool {
254
255 private static final Log LOG = LogFactory.getLog(Generator.class);
256
257
258
259
260
261
262
263
264
265
266
267
268 public static final String MULTIPLE_UNEVEN_COLUMNFAMILIES_KEY =
269 "generator.multiple.columnfamilies";
270
271 public static enum Counts {
272 SUCCESS, TERMINATING, UNDEFINED, IOEXCEPTION
273 }
274
275 public static final String USAGE = "Usage : " + Generator.class.getSimpleName() +
276 " <num mappers> <num nodes per map> <tmp output dir> [<width> <wrap multiplier>" +
277 " <num walker threads>] \n" +
278 "where <num nodes per map> should be a multiple of width*wrap multiplier, 25M by default \n" +
279 "walkers will verify random flushed loop during Generation.";
280
281 public Job job;
282
283 static class GeneratorInputFormat extends InputFormat<BytesWritable,NullWritable> {
284 static class GeneratorInputSplit extends InputSplit implements Writable {
285 @Override
286 public long getLength() throws IOException, InterruptedException {
287 return 1;
288 }
289 @Override
290 public String[] getLocations() throws IOException, InterruptedException {
291 return new String[0];
292 }
293 @Override
294 public void readFields(DataInput arg0) throws IOException {
295 }
296 @Override
297 public void write(DataOutput arg0) throws IOException {
298 }
299 }
300
301 static class GeneratorRecordReader extends RecordReader<BytesWritable,NullWritable> {
302 private long count;
303 private long numNodes;
304 private Random rand;
305
306 @Override
307 public void close() throws IOException {
308 }
309
310 @Override
311 public BytesWritable getCurrentKey() throws IOException, InterruptedException {
312 byte[] bytes = new byte[ROWKEY_LENGTH];
313 rand.nextBytes(bytes);
314 return new BytesWritable(bytes);
315 }
316
317 @Override
318 public NullWritable getCurrentValue() throws IOException, InterruptedException {
319 return NullWritable.get();
320 }
321
322 @Override
323 public float getProgress() throws IOException, InterruptedException {
324 return (float)(count / (double)numNodes);
325 }
326
327 @Override
328 public void initialize(InputSplit arg0, TaskAttemptContext context)
329 throws IOException, InterruptedException {
330 numNodes = context.getConfiguration().getLong(GENERATOR_NUM_ROWS_PER_MAP_KEY, 25000000);
331
332 rand = new SecureRandom();
333 }
334
335 @Override
336 public boolean nextKeyValue() throws IOException, InterruptedException {
337 return count++ < numNodes;
338 }
339
340 }
341
342 @Override
343 public RecordReader<BytesWritable,NullWritable> createRecordReader(
344 InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException {
345 GeneratorRecordReader rr = new GeneratorRecordReader();
346 rr.initialize(split, context);
347 return rr;
348 }
349
350 @Override
351 public List<InputSplit> getSplits(JobContext job) throws IOException, InterruptedException {
352 int numMappers = job.getConfiguration().getInt(GENERATOR_NUM_MAPPERS_KEY, 1);
353
354 ArrayList<InputSplit> splits = new ArrayList<InputSplit>(numMappers);
355
356 for (int i = 0; i < numMappers; i++) {
357 splits.add(new GeneratorInputSplit());
358 }
359
360 return splits;
361 }
362 }
363
364
365 static class OneFilePerMapperSFIF<K, V> extends SequenceFileInputFormat<K, V> {
366 @Override
367 protected boolean isSplitable(JobContext context, Path filename) {
368 return false;
369 }
370 }
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399 static class GeneratorMapper
400 extends Mapper<BytesWritable, NullWritable, NullWritable, NullWritable> {
401
402 byte[][] first = null;
403 byte[][] prev = null;
404 byte[][] current = null;
405 byte[] id;
406 long count = 0;
407 int i;
408 BufferedMutator mutator;
409 Connection connection;
410 long numNodes;
411 long wrap;
412 int width;
413 boolean multipleUnevenColumnFamilies;
414 byte[] tinyValue = new byte[] { 't' };
415 byte[] bigValue = null;
416 Configuration conf;
417
418 volatile boolean walkersStop;
419 int numWalkers;
420 volatile List<Long> flushedLoops = new ArrayList<>();
421 List<Thread> walkers = new ArrayList<>();
422
423 @Override
424 protected void setup(Context context) throws IOException, InterruptedException {
425 id = Bytes.toBytes("Job: "+context.getJobID() + " Task: " + context.getTaskAttemptID());
426 Configuration conf = context.getConfiguration();
427 connection = ConnectionFactory.createConnection(conf);
428 instantiateHTable();
429 this.width = context.getConfiguration().getInt(GENERATOR_WIDTH_KEY, WIDTH_DEFAULT);
430 current = new byte[this.width][];
431 int wrapMultiplier = context.getConfiguration().getInt(GENERATOR_WRAP_KEY, WRAP_DEFAULT);
432 this.wrap = (long)wrapMultiplier * width;
433 this.numNodes = context.getConfiguration().getLong(
434 GENERATOR_NUM_ROWS_PER_MAP_KEY, (long)WIDTH_DEFAULT * WRAP_DEFAULT);
435 if (this.numNodes < this.wrap) {
436 this.wrap = this.numNodes;
437 }
438 this.multipleUnevenColumnFamilies = isMultiUnevenColumnFamilies(context.getConfiguration());
439 this.numWalkers = context.getConfiguration().getInt(CONCURRENT_WALKER_KEY, CONCURRENT_WALKER_DEFAULT);
440 this.walkersStop = false;
441 this.conf = context.getConfiguration();
442 }
443
444 protected void instantiateHTable() throws IOException {
445 mutator = connection.getBufferedMutator(
446 new BufferedMutatorParams(getTableName(connection.getConfiguration()))
447 .writeBufferSize(4 * 1024 * 1024));
448 }
449
450 @Override
451 protected void cleanup(Context context) throws IOException ,InterruptedException {
452 joinWalkers();
453 mutator.close();
454 connection.close();
455 }
456
457 @Override
458 protected void map(BytesWritable key, NullWritable value, Context output) throws IOException {
459 current[i] = new byte[key.getLength()];
460 System.arraycopy(key.getBytes(), 0, current[i], 0, key.getLength());
461 if (++i == current.length) {
462 LOG.info("Persisting current.length=" + current.length + ", count=" + count + ", id=" +
463 Bytes.toStringBinary(id) + ", current=" + Bytes.toStringBinary(current[0]) +
464 ", i=" + i);
465 persist(output, count, prev, current, id);
466 i = 0;
467
468 if (first == null) {
469 first = current;
470 }
471 prev = current;
472 current = new byte[this.width][];
473
474 count += current.length;
475 output.setStatus("Count " + count);
476
477 if (count % wrap == 0) {
478
479
480 circularLeftShift(first);
481 persist(output, -1, prev, first, null);
482
483
484 if (numWalkers > 0) {
485 addFlushed(key.getBytes());
486 if (walkers.isEmpty()) {
487 startWalkers(numWalkers, conf, output);
488 }
489 }
490 first = null;
491 prev = null;
492 }
493 }
494 }
495
496 private static <T> void circularLeftShift(T[] first) {
497 T ez = first[0];
498 System.arraycopy(first, 1, first, 0, first.length - 1);
499 first[first.length - 1] = ez;
500 }
501
502 private void addFlushed(byte[] rowKey) {
503 synchronized (flushedLoops) {
504 flushedLoops.add(Bytes.toLong(rowKey));
505 flushedLoops.notifyAll();
506 }
507 }
508
509 protected void persist(Context output, long count, byte[][] prev, byte[][] current, byte[] id)
510 throws IOException {
511 for (int i = 0; i < current.length; i++) {
512
513 if (i % 100 == 0) {
514
515 output.progress();
516 }
517
518 Put put = new Put(current[i]);
519 put.addColumn(FAMILY_NAME, COLUMN_PREV, prev == null ? NO_KEY : prev[i]);
520
521 if (count >= 0) {
522 put.addColumn(FAMILY_NAME, COLUMN_COUNT, Bytes.toBytes(count + i));
523 }
524 if (id != null) {
525 put.addColumn(FAMILY_NAME, COLUMN_CLIENT, id);
526 }
527
528 if (this.multipleUnevenColumnFamilies) {
529
530 put.addColumn(TINY_FAMILY_NAME, TINY_FAMILY_NAME, this.tinyValue);
531
532 if (this.bigValue == null) {
533 this.bigValue = new byte[current[i].length * 10];
534 ThreadLocalRandom.current().nextBytes(this.bigValue);
535 }
536
537 put.addColumn(BIG_FAMILY_NAME, BIG_FAMILY_NAME, this.bigValue);
538 }
539 mutator.mutate(put);
540 }
541
542 mutator.flush();
543 }
544
545 private void startWalkers(int numWalkers, Configuration conf, Context context) {
546 LOG.info("Starting " + numWalkers + " concurrent walkers");
547 for (int i = 0; i < numWalkers; i++) {
548 Thread walker = new Thread(new ContinuousConcurrentWalker(conf, context));
549 walker.start();
550 walkers.add(walker);
551 }
552 }
553
554 private void joinWalkers() {
555 walkersStop = true;
556 synchronized (flushedLoops) {
557 flushedLoops.notifyAll();
558 }
559 for (Thread walker : walkers) {
560 try {
561 walker.join();
562 } catch (InterruptedException e) {
563
564 }
565 }
566 }
567
568
569
570
571
572
573
574 public class ContinuousConcurrentWalker implements Runnable {
575
576 ConcurrentWalker walker;
577 Configuration conf;
578 Context context;
579 Random rand;
580
581 public ContinuousConcurrentWalker(Configuration conf, Context context) {
582 this.conf = conf;
583 this.context = context;
584 rand = new Random();
585 }
586
587 @Override
588 public void run() {
589 while (!walkersStop) {
590 try {
591 long node = selectLoop();
592 try {
593 walkLoop(node);
594 } catch (IOException e) {
595 context.getCounter(Counts.IOEXCEPTION).increment(1l);
596 return;
597 }
598 } catch (InterruptedException e) {
599 return;
600 }
601 }
602 }
603
604 private void walkLoop(long node) throws IOException {
605 walker = new ConcurrentWalker(context);
606 walker.setConf(conf);
607 walker.run(node, wrap);
608 }
609
610 private long selectLoop () throws InterruptedException{
611 synchronized (flushedLoops) {
612 while (flushedLoops.isEmpty() && !walkersStop) {
613 flushedLoops.wait();
614 }
615 if (walkersStop) {
616 throw new InterruptedException();
617 }
618 return flushedLoops.get(rand.nextInt(flushedLoops.size()));
619 }
620 }
621 }
622
623 public static class ConcurrentWalker extends WalkerBase {
624
625 Context context;
626
627 public ConcurrentWalker(Context context) {this.context = context;}
628
629 public void run(long startKeyIn, long maxQueriesIn) throws IOException {
630
631 long maxQueries = maxQueriesIn > 0 ? maxQueriesIn : Long.MAX_VALUE;
632 byte[] startKey = Bytes.toBytes(startKeyIn);
633
634 Connection connection = ConnectionFactory.createConnection(getConf());
635 Table table = connection.getTable(getTableName(getConf()));
636 long numQueries = 0;
637
638
639
640
641 CINode node = findStartNode(table, startKey);
642 if (node == null) {
643 LOG.error("Start node not found: " + Bytes.toStringBinary(startKey));
644 throw new IOException("Start node not found: " + startKeyIn);
645 }
646 while (numQueries < maxQueries) {
647 numQueries++;
648 byte[] prev = node.prev;
649 long t1 = System.currentTimeMillis();
650 node = getNode(prev, table, node);
651 long t2 = System.currentTimeMillis();
652 if (node == null) {
653 LOG.error("ConcurrentWalker found UNDEFINED NODE: " + Bytes.toStringBinary(prev));
654 context.getCounter(Counts.UNDEFINED).increment(1l);
655 } else if (node.prev.length == NO_KEY.length) {
656 LOG.error("ConcurrentWalker found TERMINATING NODE: " +
657 Bytes.toStringBinary(node.key));
658 context.getCounter(Counts.TERMINATING).increment(1l);
659 } else {
660
661 context.getCounter(Counts.SUCCESS).increment(1l);
662 }
663 }
664 table.close();
665 connection.close();
666 }
667 }
668 }
669
670 @Override
671 public int run(String[] args) throws Exception {
672 if (args.length < 3) {
673 System.err.println(USAGE);
674 return 1;
675 }
676 try {
677 int numMappers = Integer.parseInt(args[0]);
678 long numNodes = Long.parseLong(args[1]);
679 Path tmpOutput = new Path(args[2]);
680 Integer width = (args.length < 4) ? null : Integer.parseInt(args[3]);
681 Integer wrapMultiplier = (args.length < 5) ? null : Integer.parseInt(args[4]);
682 Integer numWalkers = (args.length < 6) ? null : Integer.parseInt(args[5]);
683 return run(numMappers, numNodes, tmpOutput, width, wrapMultiplier, numWalkers);
684 } catch (NumberFormatException e) {
685 System.err.println("Parsing generator arguments failed: " + e.getMessage());
686 System.err.println(USAGE);
687 return 1;
688 }
689 }
690
691 protected void createSchema() throws IOException {
692 Configuration conf = getConf();
693 TableName tableName = getTableName(conf);
694 try (Connection conn = ConnectionFactory.createConnection(conf);
695 Admin admin = conn.getAdmin()) {
696 if (!admin.tableExists(tableName)) {
697 HTableDescriptor htd = new HTableDescriptor(getTableName(getConf()));
698 htd.addFamily(new HColumnDescriptor(FAMILY_NAME));
699
700 htd.addFamily(new HColumnDescriptor(BIG_FAMILY_NAME));
701 htd.addFamily(new HColumnDescriptor(TINY_FAMILY_NAME));
702
703
704 if (conf.getBoolean(HBaseTestingUtility.PRESPLIT_TEST_TABLE_KEY,
705 HBaseTestingUtility.PRESPLIT_TEST_TABLE)) {
706 int numberOfServers = admin.getClusterStatus().getServers().size();
707 if (numberOfServers == 0) {
708 throw new IllegalStateException("No live regionservers");
709 }
710 int regionsPerServer = conf.getInt(HBaseTestingUtility.REGIONS_PER_SERVER_KEY,
711 HBaseTestingUtility.DEFAULT_REGIONS_PER_SERVER);
712 int totalNumberOfRegions = numberOfServers * regionsPerServer;
713 LOG.info("Number of live regionservers: " + numberOfServers + ", " +
714 "pre-splitting table into " + totalNumberOfRegions + " regions " +
715 "(default regions per server: " + regionsPerServer + ")");
716
717
718 byte[][] splits = new RegionSplitter.UniformSplit().split(totalNumberOfRegions);
719
720 admin.createTable(htd, splits);
721 } else {
722
723
724
725 admin.createTable(htd);
726 }
727 }
728 } catch (MasterNotRunningException e) {
729 LOG.error("Master not running", e);
730 throw new IOException(e);
731 }
732 }
733
734 public int runRandomInputGenerator(int numMappers, long numNodes, Path tmpOutput,
735 Integer width, Integer wrapMultiplier, Integer numWalkers)
736 throws Exception {
737 LOG.info("Running RandomInputGenerator with numMappers=" + numMappers
738 + ", numNodes=" + numNodes);
739 Job job = Job.getInstance(getConf());
740
741 job.setJobName("Random Input Generator");
742 job.setNumReduceTasks(0);
743 job.setJarByClass(getClass());
744
745 job.setInputFormatClass(GeneratorInputFormat.class);
746 job.setOutputKeyClass(BytesWritable.class);
747 job.setOutputValueClass(NullWritable.class);
748
749 setJobConf(job, numMappers, numNodes, width, wrapMultiplier, numWalkers);
750
751 job.setMapperClass(Mapper.class);
752
753 FileOutputFormat.setOutputPath(job, tmpOutput);
754 job.setOutputFormatClass(SequenceFileOutputFormat.class);
755
756 boolean success = jobCompletion(job);
757
758 return success ? 0 : 1;
759 }
760
761 public int runGenerator(int numMappers, long numNodes, Path tmpOutput,
762 Integer width, Integer wrapMultiplier, Integer numWalkers)
763 throws Exception {
764 LOG.info("Running Generator with numMappers=" + numMappers +", numNodes=" + numNodes);
765 createSchema();
766 job = Job.getInstance(getConf());
767
768 job.setJobName("Link Generator");
769 job.setNumReduceTasks(0);
770 job.setJarByClass(getClass());
771
772 FileInputFormat.setInputPaths(job, tmpOutput);
773 job.setInputFormatClass(OneFilePerMapperSFIF.class);
774 job.setOutputKeyClass(NullWritable.class);
775 job.setOutputValueClass(NullWritable.class);
776
777 setJobConf(job, numMappers, numNodes, width, wrapMultiplier, numWalkers);
778
779 setMapperForGenerator(job);
780
781 job.setOutputFormatClass(NullOutputFormat.class);
782
783 job.getConfiguration().setBoolean("mapreduce.map.speculative", false);
784 TableMapReduceUtil.addDependencyJars(job);
785 TableMapReduceUtil.addDependencyJarsForClasses(job.getConfiguration(),
786 AbstractHBaseTool.class);
787 TableMapReduceUtil.initCredentials(job);
788
789 boolean success = jobCompletion(job);
790
791 return success ? 0 : 1;
792 }
793
794 protected boolean jobCompletion(Job job) throws IOException, InterruptedException,
795 ClassNotFoundException {
796 boolean success = job.waitForCompletion(true);
797 return success;
798 }
799
800 protected void setMapperForGenerator(Job job) {
801 job.setMapperClass(GeneratorMapper.class);
802 }
803
804 public int run(int numMappers, long numNodes, Path tmpOutput,
805 Integer width, Integer wrapMultiplier, Integer numWalkers)
806 throws Exception {
807 int ret = runRandomInputGenerator(numMappers, numNodes, tmpOutput, width, wrapMultiplier,
808 numWalkers);
809 if (ret > 0) {
810 return ret;
811 }
812 return runGenerator(numMappers, numNodes, tmpOutput, width, wrapMultiplier, numWalkers);
813 }
814
815 public boolean verify() {
816 try {
817 Counters counters = job.getCounters();
818
819 if (counters.findCounter(Counts.TERMINATING).getValue() > 0 ||
820 counters.findCounter(Counts.UNDEFINED).getValue() > 0 ||
821 counters.findCounter(Counts.IOEXCEPTION).getValue() > 0) {
822 LOG.error("Concurrent walker failed to verify during Generation phase");
823 LOG.error("TERMINATING nodes: " + counters.findCounter(Counts.TERMINATING).getValue());
824 LOG.error("UNDEFINED nodes: " + counters.findCounter(Counts.UNDEFINED).getValue());
825 LOG.error("IOEXCEPTION nodes: " + counters.findCounter(Counts.IOEXCEPTION).getValue());
826 return false;
827 }
828 } catch (IOException e) {
829 LOG.info("Generator verification could not find counter");
830 return false;
831 }
832 return true;
833 }
834 }
835
836
837
838
839
840
841
842 static class Search extends Configured implements Tool {
843 private static final Log LOG = LogFactory.getLog(Search.class);
844 protected Job job;
845
846 private static void printUsage(final String error) {
847 if (error != null && error.length() > 0) System.out.println("ERROR: " + error);
848 System.err.println("Usage: search <KEYS_DIR> [<MAPPERS_COUNT>]");
849 }
850
851 @Override
852 public int run(String[] args) throws Exception {
853 if (args.length < 1 || args.length > 2) {
854 printUsage(null);
855 return 1;
856 }
857 Path inputDir = new Path(args[0]);
858 int numMappers = 1;
859 if (args.length > 1) {
860 numMappers = Integer.parseInt(args[1]);
861 }
862 return run(inputDir, numMappers);
863 }
864
865
866
867
868 public static class WALSearcher extends WALPlayer {
869 public WALSearcher(Configuration conf) {
870 super(conf);
871 }
872
873
874
875
876 public static class WALMapperSearcher extends WALMapper {
877 private SortedSet<byte []> keysToFind;
878 private AtomicInteger rows = new AtomicInteger(0);
879
880 @Override
881 public void setup(Mapper<WALKey, WALEdit, ImmutableBytesWritable, Mutation>.Context context)
882 throws IOException {
883 super.setup(context);
884 try {
885 this.keysToFind = readKeysToSearch(context.getConfiguration());
886 LOG.info("Loaded keys to find: count=" + this.keysToFind.size());
887 } catch (InterruptedException e) {
888 throw new InterruptedIOException(e.toString());
889 }
890 }
891
892 @Override
893 protected boolean filter(Context context, Cell cell) {
894
895 byte [] row = new byte [cell.getRowLength()];
896 System.arraycopy(cell.getRowArray(), cell.getRowOffset(), row, 0, cell.getRowLength());
897 boolean b = this.keysToFind.contains(row);
898 if (b) {
899 String keyStr = Bytes.toStringBinary(row);
900 try {
901 LOG.info("Found cell=" + cell + " , walKey=" + context.getCurrentKey());
902 } catch (IOException|InterruptedException e) {
903 LOG.warn(e);
904 }
905 if (rows.addAndGet(1) < MISSING_ROWS_TO_LOG) {
906 context.getCounter(FOUND_GROUP_KEY, keyStr).increment(1);
907 }
908 context.getCounter(FOUND_GROUP_KEY, "CELL_WITH_MISSING_ROW").increment(1);
909 }
910 return b;
911 }
912 }
913
914
915 @Override
916 public Job createSubmittableJob(String[] args) throws IOException {
917 Job job = super.createSubmittableJob(args);
918
919 job.setJarByClass(WALMapperSearcher.class);
920 job.setMapperClass(WALMapperSearcher.class);
921 job.setOutputFormatClass(NullOutputFormat.class);
922 return job;
923 }
924 }
925
926 static final String FOUND_GROUP_KEY = "Found";
927 static final String SEARCHER_INPUTDIR_KEY = "searcher.keys.inputdir";
928
929 public int run(Path inputDir, int numMappers) throws Exception {
930 getConf().set(SEARCHER_INPUTDIR_KEY, inputDir.toString());
931 SortedSet<byte []> keys = readKeysToSearch(getConf());
932 if (keys.isEmpty()) throw new RuntimeException("No keys to find");
933 LOG.info("Count of keys to find: " + keys.size());
934 for(byte [] key: keys) LOG.info("Key: " + Bytes.toStringBinary(key));
935 Path hbaseDir = new Path(getConf().get(HConstants.HBASE_DIR));
936
937 Path walsDir = new Path(hbaseDir, HConstants.HREGION_LOGDIR_NAME);
938 Path oldWalsDir = new Path(hbaseDir, HConstants.HREGION_OLDLOGDIR_NAME);
939 LOG.info("Running Search with keys inputDir=" + inputDir +", numMappers=" + numMappers +
940 " against " + getConf().get(HConstants.HBASE_DIR));
941 int ret = ToolRunner.run(new WALSearcher(getConf()), new String [] {walsDir.toString(), ""});
942 if (ret != 0) return ret;
943 return ToolRunner.run(new WALSearcher(getConf()), new String [] {oldWalsDir.toString(), ""});
944 }
945
946 static SortedSet<byte []> readKeysToSearch(final Configuration conf)
947 throws IOException, InterruptedException {
948 Path keysInputDir = new Path(conf.get(SEARCHER_INPUTDIR_KEY));
949 FileSystem fs = FileSystem.get(conf);
950 SortedSet<byte []> result = new TreeSet<byte []>(Bytes.BYTES_COMPARATOR);
951 if (!fs.exists(keysInputDir)) {
952 throw new FileNotFoundException(keysInputDir.toString());
953 }
954 if (!fs.isDirectory(keysInputDir)) {
955 throw new UnsupportedOperationException("TODO");
956 } else {
957 RemoteIterator<LocatedFileStatus> iterator = fs.listFiles(keysInputDir, false);
958 while(iterator.hasNext()) {
959 LocatedFileStatus keyFileStatus = iterator.next();
960
961 if (keyFileStatus.getPath().getName().startsWith("_")) continue;
962 result.addAll(readFileToSearch(conf, fs, keyFileStatus));
963 }
964 }
965 return result;
966 }
967
968 private static SortedSet<byte[]> readFileToSearch(final Configuration conf,
969 final FileSystem fs, final LocatedFileStatus keyFileStatus) throws IOException,
970 InterruptedException {
971 SortedSet<byte []> result = new TreeSet<byte []>(Bytes.BYTES_COMPARATOR);
972
973
974 TaskAttemptContext context = new TaskAttemptContextImpl(conf, new TaskAttemptID());
975 try (SequenceFileAsBinaryInputFormat.SequenceFileAsBinaryRecordReader rr =
976 new SequenceFileAsBinaryInputFormat.SequenceFileAsBinaryRecordReader()) {
977 InputSplit is =
978 new FileSplit(keyFileStatus.getPath(), 0, keyFileStatus.getLen(), new String [] {});
979 rr.initialize(is, context);
980 while (rr.nextKeyValue()) {
981 rr.getCurrentKey();
982 BytesWritable bw = rr.getCurrentValue();
983 if (Verify.VerifyReducer.whichType(bw.getBytes()) == Verify.Counts.UNDEFINED) {
984 byte[] key = new byte[rr.getCurrentKey().getLength()];
985 System.arraycopy(rr.getCurrentKey().getBytes(), 0, key, 0, rr.getCurrentKey()
986 .getLength());
987 result.add(key);
988 }
989 }
990 }
991 return result;
992 }
993 }
994
995
996
997
998
999 static class Verify extends Configured implements Tool {
1000
1001 private static final Log LOG = LogFactory.getLog(Verify.class);
1002 protected static final BytesWritable DEF = new BytesWritable(new byte[] { 0 });
1003 protected static final BytesWritable DEF_LOST_FAMILIES = new BytesWritable(new byte[] { 1 });
1004
1005 protected Job job;
1006
1007 public static class VerifyMapper extends TableMapper<BytesWritable, BytesWritable> {
1008 private BytesWritable row = new BytesWritable();
1009 private BytesWritable ref = new BytesWritable();
1010
1011 private boolean multipleUnevenColumnFamilies;
1012
1013 @Override
1014 protected void setup(
1015 Mapper<ImmutableBytesWritable, Result, BytesWritable, BytesWritable>.Context context)
1016 throws IOException, InterruptedException {
1017 this.multipleUnevenColumnFamilies = isMultiUnevenColumnFamilies(context.getConfiguration());
1018 }
1019
1020 @Override
1021 protected void map(ImmutableBytesWritable key, Result value, Context context)
1022 throws IOException ,InterruptedException {
1023 byte[] rowKey = key.get();
1024 row.set(rowKey, 0, rowKey.length);
1025 if (multipleUnevenColumnFamilies
1026 && (!value.containsColumn(BIG_FAMILY_NAME, BIG_FAMILY_NAME) || !value.containsColumn(
1027 TINY_FAMILY_NAME, TINY_FAMILY_NAME))) {
1028 context.write(row, DEF_LOST_FAMILIES);
1029 } else {
1030 context.write(row, DEF);
1031 }
1032 byte[] prev = value.getValue(FAMILY_NAME, COLUMN_PREV);
1033 if (prev != null && prev.length > 0) {
1034 ref.set(prev, 0, prev.length);
1035 context.write(ref, row);
1036 } else {
1037 LOG.warn(String.format("Prev is not set for: %s", Bytes.toStringBinary(rowKey)));
1038 }
1039 }
1040 }
1041
1042
1043
1044
1045
1046 public static enum Counts {
1047 UNREFERENCED, UNDEFINED, REFERENCED, CORRUPT, EXTRAREFERENCES, EXTRA_UNDEF_REFERENCES,
1048 LOST_FAMILIES
1049 }
1050
1051
1052
1053
1054
1055
1056 public static class VerifyReducer extends
1057 Reducer<BytesWritable, BytesWritable, BytesWritable, BytesWritable> {
1058 private ArrayList<byte[]> refs = new ArrayList<byte[]>();
1059 private final BytesWritable UNREF = new BytesWritable(addPrefixFlag(
1060 Counts.UNREFERENCED.ordinal(), new byte[] {}));
1061 private final BytesWritable LOSTFAM = new BytesWritable(addPrefixFlag(
1062 Counts.LOST_FAMILIES.ordinal(), new byte[] {}));
1063
1064 private AtomicInteger rows = new AtomicInteger(0);
1065 private Connection connection;
1066
1067 @Override
1068 protected void setup(Reducer<BytesWritable, BytesWritable, BytesWritable, BytesWritable>.Context context)
1069 throws IOException, InterruptedException {
1070 super.setup(context);
1071 this.connection = ConnectionFactory.createConnection(context.getConfiguration());
1072 }
1073
1074 @Override
1075 protected void cleanup(
1076 Reducer<BytesWritable, BytesWritable, BytesWritable, BytesWritable>.Context context)
1077 throws IOException, InterruptedException {
1078 if (this.connection != null) {
1079 this.connection.close();
1080 }
1081 super.cleanup(context);
1082 }
1083
1084
1085
1086
1087
1088
1089
1090 public static byte[] addPrefixFlag(final int ordinal, final byte [] r) {
1091 byte[] prefix = Bytes.toBytes((short)ordinal);
1092 if (prefix.length != Bytes.SIZEOF_SHORT) {
1093 throw new RuntimeException("Unexpected size: " + prefix.length);
1094 }
1095 byte[] result = new byte[prefix.length + r.length];
1096 System.arraycopy(prefix, 0, result, 0, prefix.length);
1097 System.arraycopy(r, 0, result, prefix.length, r.length);
1098 return result;
1099 }
1100
1101
1102
1103
1104
1105
1106 public static Counts whichType(final byte [] bs) {
1107 int ordinal = Bytes.toShort(bs, 0, Bytes.SIZEOF_SHORT);
1108 return Counts.values()[ordinal];
1109 }
1110
1111
1112
1113
1114
1115 public static byte[] getRowOnly(BytesWritable bw) {
1116 byte[] bytes = new byte [bw.getLength() - Bytes.SIZEOF_SHORT];
1117 System.arraycopy(bw.getBytes(), Bytes.SIZEOF_SHORT, bytes, 0, bytes.length);
1118 return bytes;
1119 }
1120
1121 @Override
1122 public void reduce(BytesWritable key, Iterable<BytesWritable> values, Context context)
1123 throws IOException, InterruptedException {
1124 int defCount = 0;
1125 boolean lostFamilies = false;
1126 refs.clear();
1127 for (BytesWritable type : values) {
1128 if (type.getLength() == DEF.getLength()) {
1129 defCount++;
1130 if (type.getBytes()[0] == 1) {
1131 lostFamilies = true;
1132 }
1133 } else {
1134 byte[] bytes = new byte[type.getLength()];
1135 System.arraycopy(type.getBytes(), 0, bytes, 0, type.getLength());
1136 refs.add(bytes);
1137 }
1138 }
1139
1140
1141 StringBuilder refsSb = null;
1142 String keyString = Bytes.toStringBinary(key.getBytes(), 0, key.getLength());
1143 if (defCount == 0 || refs.size() != 1) {
1144 refsSb = dumpExtraInfoOnRefs(key, context, refs);
1145 LOG.error("LinkedListError: key=" + keyString + ", reference(s)=" +
1146 (refsSb != null? refsSb.toString(): ""));
1147 }
1148 if (lostFamilies) {
1149 LOG.error("LinkedListError: key=" + keyString + ", lost big or tiny families");
1150 context.getCounter(Counts.LOST_FAMILIES).increment(1);
1151 context.write(key, LOSTFAM);
1152 }
1153
1154 if (defCount == 0 && refs.size() > 0) {
1155
1156
1157
1158 for (int i = 0; i < refs.size(); i++) {
1159 byte[] bs = refs.get(i);
1160 int ordinal;
1161 if (i <= 0) {
1162 ordinal = Counts.UNDEFINED.ordinal();
1163 context.write(key, new BytesWritable(addPrefixFlag(ordinal, bs)));
1164 context.getCounter(Counts.UNDEFINED).increment(1);
1165 } else {
1166 ordinal = Counts.EXTRA_UNDEF_REFERENCES.ordinal();
1167 context.write(key, new BytesWritable(addPrefixFlag(ordinal, bs)));
1168 }
1169 }
1170 if (rows.addAndGet(1) < MISSING_ROWS_TO_LOG) {
1171
1172
1173
1174
1175 context.getCounter("undef", keyString).increment(1);
1176 }
1177 } else if (defCount > 0 && refs.size() == 0) {
1178
1179 context.write(key, UNREF);
1180 context.getCounter(Counts.UNREFERENCED).increment(1);
1181 if (rows.addAndGet(1) < MISSING_ROWS_TO_LOG) {
1182 context.getCounter("unref", keyString).increment(1);
1183 }
1184 } else {
1185 if (refs.size() > 1) {
1186
1187 for (int i = 1; i < refs.size(); i++) {
1188 context.write(key,
1189 new BytesWritable(addPrefixFlag(Counts.EXTRAREFERENCES.ordinal(), refs.get(i))));
1190 }
1191 context.getCounter(Counts.EXTRAREFERENCES).increment(refs.size() - 1);
1192 }
1193
1194 context.getCounter(Counts.REFERENCED).increment(1);
1195 }
1196 }
1197
1198
1199
1200
1201
1202
1203 private StringBuilder dumpExtraInfoOnRefs(final BytesWritable key, final Context context,
1204 final List<byte []> refs)
1205 throws IOException {
1206 StringBuilder refsSb = null;
1207 if (refs.isEmpty()) return refsSb;
1208 refsSb = new StringBuilder();
1209 String comma = "";
1210
1211
1212
1213 TableName tn = getTableName(context.getConfiguration());
1214 try (Table t = this.connection.getTable(tn)) {
1215 for (byte [] ref : refs) {
1216 Result r = t.get(new Get(ref));
1217 List<Cell> cells = r.listCells();
1218 String ts = (cells != null && !cells.isEmpty())?
1219 new java.util.Date(cells.get(0).getTimestamp()).toString(): "";
1220 byte [] b = r.getValue(FAMILY_NAME, COLUMN_CLIENT);
1221 String jobStr = (b != null && b.length > 0)? Bytes.toString(b): "";
1222 b = r.getValue(FAMILY_NAME, COLUMN_COUNT);
1223 long count = (b != null && b.length > 0)? Bytes.toLong(b): -1;
1224 b = r.getValue(FAMILY_NAME, COLUMN_PREV);
1225 String refRegionLocation = "";
1226 String keyRegionLocation = "";
1227 if (b != null && b.length > 0) {
1228 try (RegionLocator rl = this.connection.getRegionLocator(tn)) {
1229 HRegionLocation hrl = rl.getRegionLocation(b);
1230 if (hrl != null) refRegionLocation = hrl.toString();
1231
1232 hrl = rl.getRegionLocation(key.getBytes());
1233 if (hrl != null) keyRegionLocation = hrl.toString();
1234 }
1235 }
1236 LOG.error("Extras on ref without a def, ref=" + Bytes.toStringBinary(ref) +
1237 ", refPrevEqualsKey=" +
1238 (Bytes.compareTo(key.getBytes(), 0, key.getLength(), b, 0, b.length) == 0) +
1239 ", key=" + Bytes.toStringBinary(key.getBytes(), 0, key.getLength()) +
1240 ", ref row date=" + ts + ", jobStr=" + jobStr +
1241 ", ref row count=" + count +
1242 ", ref row regionLocation=" + refRegionLocation +
1243 ", key row regionLocation=" + keyRegionLocation);
1244 refsSb.append(comma);
1245 comma = ",";
1246 refsSb.append(Bytes.toStringBinary(ref));
1247 }
1248 }
1249 return refsSb;
1250 }
1251 }
1252
1253 @Override
1254 public int run(String[] args) throws Exception {
1255 if (args.length != 2) {
1256 System.out.println("Usage : " + Verify.class.getSimpleName()
1257 + " <output dir> <num reducers>");
1258 return 0;
1259 }
1260
1261 String outputDir = args[0];
1262 int numReducers = Integer.parseInt(args[1]);
1263
1264 return run(outputDir, numReducers);
1265 }
1266
1267 public int run(String outputDir, int numReducers) throws Exception {
1268 return run(new Path(outputDir), numReducers);
1269 }
1270
1271 public int run(Path outputDir, int numReducers) throws Exception {
1272 LOG.info("Running Verify with outputDir=" + outputDir +", numReducers=" + numReducers);
1273
1274 job = Job.getInstance(getConf());
1275
1276 job.setJobName("Link Verifier");
1277 job.setNumReduceTasks(numReducers);
1278 job.setJarByClass(getClass());
1279
1280 setJobScannerConf(job);
1281
1282 Scan scan = new Scan();
1283 scan.addColumn(FAMILY_NAME, COLUMN_PREV);
1284 scan.setCaching(10000);
1285 scan.setCacheBlocks(false);
1286 if (isMultiUnevenColumnFamilies(getConf())) {
1287 scan.addColumn(BIG_FAMILY_NAME, BIG_FAMILY_NAME);
1288 scan.addColumn(TINY_FAMILY_NAME, TINY_FAMILY_NAME);
1289 }
1290
1291 TableMapReduceUtil.initTableMapperJob(getTableName(getConf()).getName(), scan,
1292 VerifyMapper.class, BytesWritable.class, BytesWritable.class, job);
1293 TableMapReduceUtil.addDependencyJarsForClasses(job.getConfiguration(),
1294 AbstractHBaseTool.class);
1295
1296 job.getConfiguration().setBoolean("mapreduce.map.speculative", false);
1297
1298 job.setReducerClass(VerifyReducer.class);
1299 job.setOutputFormatClass(SequenceFileAsBinaryOutputFormat.class);
1300 job.setOutputKeyClass(BytesWritable.class);
1301 job.setOutputValueClass(BytesWritable.class);
1302 TextOutputFormat.setOutputPath(job, outputDir);
1303
1304 boolean success = job.waitForCompletion(true);
1305
1306 if (success) {
1307 Counters counters = job.getCounters();
1308 if (null == counters) {
1309 LOG.warn("Counters were null, cannot verify Job completion");
1310
1311 return 0;
1312 }
1313
1314
1315 if (verifyUnexpectedValues(counters)) {
1316
1317 return 0;
1318 }
1319 }
1320
1321
1322 return 1;
1323 }
1324
1325 public boolean verify(long expectedReferenced) throws Exception {
1326 if (job == null) {
1327 throw new IllegalStateException("You should call run() first");
1328 }
1329
1330 Counters counters = job.getCounters();
1331
1332
1333 boolean success = verifyExpectedValues(expectedReferenced, counters);
1334
1335 if (!verifyUnexpectedValues(counters)) {
1336
1337 success = false;
1338 }
1339
1340 if (!success) {
1341 handleFailure(counters);
1342 }
1343 return success;
1344 }
1345
1346
1347
1348
1349
1350
1351
1352
1353
1354
1355 protected boolean verifyExpectedValues(long expectedReferenced, Counters counters) {
1356 final Counter referenced = counters.findCounter(Counts.REFERENCED);
1357 final Counter unreferenced = counters.findCounter(Counts.UNREFERENCED);
1358 boolean success = true;
1359
1360 if (expectedReferenced != referenced.getValue()) {
1361 LOG.error("Expected referenced count does not match with actual referenced count. " +
1362 "expected referenced=" + expectedReferenced + " ,actual=" + referenced.getValue());
1363 success = false;
1364 }
1365
1366 if (unreferenced.getValue() > 0) {
1367 final Counter multiref = counters.findCounter(Counts.EXTRAREFERENCES);
1368 boolean couldBeMultiRef = (multiref.getValue() == unreferenced.getValue());
1369 LOG.error("Unreferenced nodes were not expected. Unreferenced count=" + unreferenced.getValue()
1370 + (couldBeMultiRef ? "; could be due to duplicate random numbers" : ""));
1371 success = false;
1372 }
1373
1374 return success;
1375 }
1376
1377
1378
1379
1380
1381
1382
1383
1384 protected boolean verifyUnexpectedValues(Counters counters) {
1385 final Counter undefined = counters.findCounter(Counts.UNDEFINED);
1386 final Counter lostfamilies = counters.findCounter(Counts.LOST_FAMILIES);
1387 boolean success = true;
1388
1389 if (undefined.getValue() > 0) {
1390 LOG.error("Found an undefined node. Undefined count=" + undefined.getValue());
1391 success = false;
1392 }
1393
1394 if (lostfamilies.getValue() > 0) {
1395 LOG.error("Found nodes which lost big or tiny families, count=" + lostfamilies.getValue());
1396 success = false;
1397 }
1398
1399 return success;
1400 }
1401
1402 protected void handleFailure(Counters counters) throws IOException {
1403 Configuration conf = job.getConfiguration();
1404 TableName tableName = getTableName(conf);
1405 try (Connection conn = ConnectionFactory.createConnection(conf)) {
1406 try (RegionLocator rl = conn.getRegionLocator(tableName)) {
1407 CounterGroup g = counters.getGroup("undef");
1408 Iterator<Counter> it = g.iterator();
1409 while (it.hasNext()) {
1410 String keyString = it.next().getName();
1411 byte[] key = Bytes.toBytes(keyString);
1412 HRegionLocation loc = rl.getRegionLocation(key, true);
1413 LOG.error("undefined row " + keyString + ", " + loc);
1414 }
1415 g = counters.getGroup("unref");
1416 it = g.iterator();
1417 while (it.hasNext()) {
1418 String keyString = it.next().getName();
1419 byte[] key = Bytes.toBytes(keyString);
1420 HRegionLocation loc = rl.getRegionLocation(key, true);
1421 LOG.error("unreferred row " + keyString + ", " + loc);
1422 }
1423 }
1424 }
1425 }
1426 }
1427
1428
1429
1430
1431
1432 static class Loop extends Configured implements Tool {
1433
1434 private static final Log LOG = LogFactory.getLog(Loop.class);
1435 private static final String USAGE = "Usage: Loop <num iterations> <num mappers> " +
1436 "<num nodes per mapper> <output dir> <num reducers> [<width> <wrap multiplier>" +
1437 " <num walker threads>] \n" +
1438 "where <num nodes per map> should be a multiple of width*wrap multiplier, 25M by default \n" +
1439 "walkers will select and verify random flushed loop during Generation.";
1440
1441 IntegrationTestBigLinkedList it;
1442
1443 protected void runGenerator(int numMappers, long numNodes,
1444 String outputDir, Integer width, Integer wrapMultiplier, Integer numWalkers)
1445 throws Exception {
1446 Path outputPath = new Path(outputDir);
1447 UUID uuid = UUID.randomUUID();
1448 Path generatorOutput = new Path(outputPath, uuid.toString());
1449
1450 Generator generator = new Generator();
1451 generator.setConf(getConf());
1452 int retCode = generator.run(numMappers, numNodes, generatorOutput, width, wrapMultiplier,
1453 numWalkers);
1454 if (retCode > 0) {
1455 throw new RuntimeException("Generator failed with return code: " + retCode);
1456 }
1457 if (numWalkers > 0) {
1458 if (!generator.verify()) {
1459 throw new RuntimeException("Generator.verify failed");
1460 }
1461 }
1462 }
1463
1464 protected void runVerify(String outputDir,
1465 int numReducers, long expectedNumNodes) throws Exception {
1466 Path outputPath = new Path(outputDir);
1467 UUID uuid = UUID.randomUUID();
1468 Path iterationOutput = new Path(outputPath, uuid.toString());
1469
1470 Verify verify = new Verify();
1471 verify.setConf(getConf());
1472 int retCode = verify.run(iterationOutput, numReducers);
1473 if (retCode > 0) {
1474 throw new RuntimeException("Verify.run failed with return code: " + retCode);
1475 }
1476
1477 if (!verify.verify(expectedNumNodes)) {
1478 throw new RuntimeException("Verify.verify failed");
1479 }
1480 LOG.info("Verify finished with success. Total nodes=" + expectedNumNodes);
1481 }
1482
1483 @Override
1484 public int run(String[] args) throws Exception {
1485 if (args.length < 5) {
1486 System.err.println(USAGE);
1487 return 1;
1488 }
1489 try {
1490 int numIterations = Integer.parseInt(args[0]);
1491 int numMappers = Integer.parseInt(args[1]);
1492 long numNodes = Long.parseLong(args[2]);
1493 String outputDir = args[3];
1494 int numReducers = Integer.parseInt(args[4]);
1495 Integer width = (args.length < 6) ? null : Integer.parseInt(args[5]);
1496 Integer wrapMultiplier = (args.length < 7) ? null : Integer.parseInt(args[6]);
1497 Integer numWalkers = (args.length < 8) ? 0 : Integer.parseInt(args[7]);
1498
1499 long expectedNumNodes = 0;
1500
1501 if (numIterations < 0) {
1502 numIterations = Integer.MAX_VALUE;
1503 }
1504 LOG.info("Running Loop with args:" + Arrays.deepToString(args));
1505 for (int i = 0; i < numIterations; i++) {
1506 LOG.info("Starting iteration = " + i);
1507 runGenerator(numMappers, numNodes, outputDir, width, wrapMultiplier, numWalkers);
1508 expectedNumNodes += numMappers * numNodes;
1509 runVerify(outputDir, numReducers, expectedNumNodes);
1510 }
1511 return 0;
1512 } catch (NumberFormatException e) {
1513 System.err.println("Parsing loop arguments failed: " + e.getMessage());
1514 System.err.println(USAGE);
1515 return 1;
1516 }
1517 }
1518 }
1519
1520
1521
1522
1523 private static class Print extends Configured implements Tool {
1524 @Override
1525 public int run(String[] args) throws Exception {
1526 Options options = new Options();
1527 options.addOption("s", "start", true, "start key");
1528 options.addOption("e", "end", true, "end key");
1529 options.addOption("l", "limit", true, "number to print");
1530
1531 GnuParser parser = new GnuParser();
1532 CommandLine cmd = null;
1533 try {
1534 cmd = parser.parse(options, args);
1535 if (cmd.getArgs().length != 0) {
1536 throw new ParseException("Command takes no arguments");
1537 }
1538 } catch (ParseException e) {
1539 System.err.println("Failed to parse command line " + e.getMessage());
1540 System.err.println();
1541 HelpFormatter formatter = new HelpFormatter();
1542 formatter.printHelp(getClass().getSimpleName(), options);
1543 System.exit(-1);
1544 }
1545
1546 Table table = new HTable(getConf(), getTableName(getConf()));
1547
1548 Scan scan = new Scan();
1549 scan.setBatch(10000);
1550
1551 if (cmd.hasOption("s"))
1552 scan.setStartRow(Bytes.toBytesBinary(cmd.getOptionValue("s")));
1553
1554 if (cmd.hasOption("e"))
1555 scan.setStopRow(Bytes.toBytesBinary(cmd.getOptionValue("e")));
1556
1557 int limit = 0;
1558 if (cmd.hasOption("l"))
1559 limit = Integer.parseInt(cmd.getOptionValue("l"));
1560 else
1561 limit = 100;
1562
1563 ResultScanner scanner = table.getScanner(scan);
1564
1565 CINode node = new CINode();
1566 Result result = scanner.next();
1567 int count = 0;
1568 while (result != null && count++ < limit) {
1569 node = getCINode(result, node);
1570 System.out.printf("%s:%s:%012d:%s\n", Bytes.toStringBinary(node.key),
1571 Bytes.toStringBinary(node.prev), node.count, node.client);
1572 result = scanner.next();
1573 }
1574 scanner.close();
1575 table.close();
1576
1577 return 0;
1578 }
1579 }
1580
1581
1582
1583
1584 private static class Delete extends Configured implements Tool {
1585 @Override
1586 public int run(String[] args) throws Exception {
1587 if (args.length != 1) {
1588 System.out.println("Usage : " + Delete.class.getSimpleName() + " <node to delete>");
1589 return 0;
1590 }
1591 byte[] val = Bytes.toBytesBinary(args[0]);
1592
1593 org.apache.hadoop.hbase.client.Delete delete
1594 = new org.apache.hadoop.hbase.client.Delete(val);
1595
1596 Table table = new HTable(getConf(), getTableName(getConf()));
1597 table.delete(delete);
1598 table.close();
1599
1600 System.out.println("Delete successful");
1601 return 0;
1602 }
1603 }
1604
1605 abstract static class WalkerBase extends Configured{
1606 protected static CINode findStartNode(Table table, byte[] startKey) throws IOException {
1607 Scan scan = new Scan();
1608 scan.setStartRow(startKey);
1609 scan.setBatch(1);
1610 scan.addColumn(FAMILY_NAME, COLUMN_PREV);
1611
1612 long t1 = System.currentTimeMillis();
1613 ResultScanner scanner = table.getScanner(scan);
1614 Result result = scanner.next();
1615 long t2 = System.currentTimeMillis();
1616 scanner.close();
1617
1618 if ( result != null) {
1619 CINode node = getCINode(result, new CINode());
1620 System.out.printf("FSR %d %s\n", t2 - t1, Bytes.toStringBinary(node.key));
1621 return node;
1622 }
1623
1624 System.out.println("FSR " + (t2 - t1));
1625
1626 return null;
1627 }
1628 protected CINode getNode(byte[] row, Table table, CINode node) throws IOException {
1629 Get get = new Get(row);
1630 get.addColumn(FAMILY_NAME, COLUMN_PREV);
1631 Result result = table.get(get);
1632 return getCINode(result, node);
1633 }
1634 }
1635
1636
1637
1638
1639 private static class Walker extends WalkerBase implements Tool {
1640
1641 public Walker(){}
1642
1643 @Override
1644 public int run(String[] args) throws IOException {
1645
1646 Options options = new Options();
1647 options.addOption("n", "num", true, "number of queries");
1648 options.addOption("s", "start", true, "key to start at, binary string");
1649 options.addOption("l", "logevery", true, "log every N queries");
1650
1651 GnuParser parser = new GnuParser();
1652 CommandLine cmd = null;
1653 try {
1654 cmd = parser.parse(options, args);
1655 if (cmd.getArgs().length != 0) {
1656 throw new ParseException("Command takes no arguments");
1657 }
1658 } catch (ParseException e) {
1659 System.err.println("Failed to parse command line " + e.getMessage());
1660 System.err.println();
1661 HelpFormatter formatter = new HelpFormatter();
1662 formatter.printHelp(getClass().getSimpleName(), options);
1663 System.exit(-1);
1664 }
1665
1666 long maxQueries = Long.MAX_VALUE;
1667 if (cmd.hasOption('n')) {
1668 maxQueries = Long.parseLong(cmd.getOptionValue("n"));
1669 }
1670 Random rand = new SecureRandom();
1671 boolean isSpecificStart = cmd.hasOption('s');
1672
1673 byte[] startKey = isSpecificStart ? Bytes.toBytesBinary(cmd.getOptionValue('s')) : null;
1674 int logEvery = cmd.hasOption('l') ? Integer.parseInt(cmd.getOptionValue('l')) : 1;
1675
1676 Table table = new HTable(getConf(), getTableName(getConf()));
1677 long numQueries = 0;
1678
1679
1680
1681 while (numQueries < maxQueries && (numQueries == 0 || !isSpecificStart)) {
1682 if (!isSpecificStart) {
1683 startKey = new byte[ROWKEY_LENGTH];
1684 rand.nextBytes(startKey);
1685 }
1686 CINode node = findStartNode(table, startKey);
1687 if (node == null && isSpecificStart) {
1688 System.err.printf("Start node not found: %s \n", Bytes.toStringBinary(startKey));
1689 }
1690 numQueries++;
1691 while (node != null && node.prev.length != NO_KEY.length &&
1692 numQueries < maxQueries) {
1693 byte[] prev = node.prev;
1694 long t1 = System.currentTimeMillis();
1695 node = getNode(prev, table, node);
1696 long t2 = System.currentTimeMillis();
1697 if (logEvery > 0 && numQueries % logEvery == 0) {
1698 System.out.printf("CQ %d: %d %s \n", numQueries, t2 - t1, Bytes.toStringBinary(prev));
1699 }
1700 numQueries++;
1701 if (node == null) {
1702 System.err.printf("UNDEFINED NODE %s \n", Bytes.toStringBinary(prev));
1703 } else if (node.prev.length == NO_KEY.length) {
1704 System.err.printf("TERMINATING NODE %s \n", Bytes.toStringBinary(node.key));
1705 }
1706 }
1707 }
1708 table.close();
1709 return 0;
1710 }
1711 }
1712
1713 private static class Clean extends Configured implements Tool {
1714 @Override public int run(String[] args) throws Exception {
1715 if (args.length < 1) {
1716 System.err.println("Usage: Clean <output dir>");
1717 return -1;
1718 }
1719
1720 Path p = new Path(args[0]);
1721 Configuration conf = getConf();
1722 TableName tableName = getTableName(conf);
1723 try (FileSystem fs = HFileSystem.get(conf);
1724 Connection conn = ConnectionFactory.createConnection(conf);
1725 Admin admin = conn.getAdmin()) {
1726 if (admin.tableExists(tableName)) {
1727 admin.disableTable(tableName);
1728 admin.deleteTable(tableName);
1729 }
1730
1731 if (fs.exists(p)) {
1732 fs.delete(p, true);
1733 }
1734 }
1735
1736 return 0;
1737 }
1738 }
1739
1740 static TableName getTableName(Configuration conf) {
1741 return TableName.valueOf(conf.get(TABLE_NAME_KEY, DEFAULT_TABLE_NAME));
1742 }
1743
1744 private static CINode getCINode(Result result, CINode node) {
1745 node.key = Bytes.copy(result.getRow());
1746 if (result.containsColumn(FAMILY_NAME, COLUMN_PREV)) {
1747 node.prev = Bytes.copy(result.getValue(FAMILY_NAME, COLUMN_PREV));
1748 } else {
1749 node.prev = NO_KEY;
1750 }
1751 if (result.containsColumn(FAMILY_NAME, COLUMN_COUNT)) {
1752 node.count = Bytes.toLong(result.getValue(FAMILY_NAME, COLUMN_COUNT));
1753 } else {
1754 node.count = -1;
1755 }
1756 if (result.containsColumn(FAMILY_NAME, COLUMN_CLIENT)) {
1757 node.client = Bytes.toString(result.getValue(FAMILY_NAME, COLUMN_CLIENT));
1758 } else {
1759 node.client = "";
1760 }
1761 return node;
1762 }
1763
1764 protected IntegrationTestingUtility util;
1765
1766 @Override
1767 public void setUpCluster() throws Exception {
1768 util = getTestingUtil(getConf());
1769 boolean isDistributed = util.isDistributedCluster();
1770 util.initializeCluster(isDistributed ? 1 : this.NUM_SLAVES_BASE);
1771 if (!isDistributed) {
1772 util.startMiniMapReduceCluster();
1773 }
1774 this.setConf(util.getConfiguration());
1775 }
1776
1777 @Override
1778 public void cleanUpCluster() throws Exception {
1779 super.cleanUpCluster();
1780 if (util.isDistributedCluster()) {
1781 util.shutdownMiniMapReduceCluster();
1782 }
1783 }
1784
1785 private static boolean isMultiUnevenColumnFamilies(Configuration conf) {
1786 return conf.getBoolean(Generator.MULTIPLE_UNEVEN_COLUMNFAMILIES_KEY,true);
1787 }
1788
1789 @Test
1790 public void testContinuousIngest() throws IOException, Exception {
1791
1792 Configuration conf = getTestingUtil(getConf()).getConfiguration();
1793 if (isMultiUnevenColumnFamilies(getConf())) {
1794
1795 conf.set(FlushPolicyFactory.HBASE_FLUSH_POLICY_KEY, FlushLargeStoresPolicy.class.getName());
1796 }
1797 int ret =
1798 ToolRunner.run(conf, new Loop(), new String[] { "1", "1", "2000000",
1799 util.getDataTestDirOnTestFS("IntegrationTestBigLinkedList").toString(), "1" });
1800 org.junit.Assert.assertEquals(0, ret);
1801 }
1802
1803 private void usage() {
1804 System.err.println("Usage: " + this.getClass().getSimpleName() + " COMMAND [COMMAND options]");
1805 printCommands();
1806 }
1807
1808 private void printCommands() {
1809 System.err.println("Commands:");
1810 System.err.println(" generator Map only job that generates data.");
1811 System.err.println(" verify A map reduce job that looks for holes. Check return code and");
1812 System.err.println(" look at the counts after running. See REFERENCED and");
1813 System.err.println(" UNREFERENCED are ok. Any UNDEFINED counts are bad. Do not run");
1814 System.err.println(" with the Generator.");
1815 System.err.println(" walker " +
1816 "Standalone program that starts following a linked list & emits timing info.");
1817 System.err.println(" print Standalone program that prints nodes in the linked list.");
1818 System.err.println(" delete Standalone program that deletes a·single node.");
1819 System.err.println(" loop Program to Loop through Generator and Verify steps");
1820 System.err.println(" clean Program to clean all left over detritus.");
1821 System.err.println(" search Search for missing keys.");
1822 System.err.flush();
1823 }
1824
1825 @Override
1826 protected void processOptions(CommandLine cmd) {
1827 super.processOptions(cmd);
1828 String[] args = cmd.getArgs();
1829
1830 if (args.length < 1) {
1831 printUsage(this.getClass().getSimpleName() +
1832 " <general options> COMMAND [<COMMAND options>]", "General options:", "");
1833 printCommands();
1834
1835 throw new RuntimeException("Incorrect Number of args.");
1836 }
1837 toRun = args[0];
1838 otherArgs = Arrays.copyOfRange(args, 1, args.length);
1839 }
1840
1841 @Override
1842 public int runTestFromCommandLine() throws Exception {
1843 Tool tool = null;
1844 if (toRun.equalsIgnoreCase("Generator")) {
1845 tool = new Generator();
1846 } else if (toRun.equalsIgnoreCase("Verify")) {
1847 tool = new Verify();
1848 } else if (toRun.equalsIgnoreCase("Loop")) {
1849 Loop loop = new Loop();
1850 loop.it = this;
1851 tool = loop;
1852 } else if (toRun.equalsIgnoreCase("Walker")) {
1853 tool = new Walker();
1854 } else if (toRun.equalsIgnoreCase("Print")) {
1855 tool = new Print();
1856 } else if (toRun.equalsIgnoreCase("Delete")) {
1857 tool = new Delete();
1858 } else if (toRun.equalsIgnoreCase("Clean")) {
1859 tool = new Clean();
1860 } else if (toRun.equalsIgnoreCase("Search")) {
1861 tool = new Search();
1862 } else {
1863 usage();
1864 throw new RuntimeException("Unknown arg");
1865 }
1866
1867 return ToolRunner.run(getConf(), tool, otherArgs);
1868 }
1869
1870 @Override
1871 public TableName getTablename() {
1872 Configuration c = getConf();
1873 return TableName.valueOf(c.get(TABLE_NAME_KEY, DEFAULT_TABLE_NAME));
1874 }
1875
1876 @Override
1877 protected Set<String> getColumnFamilies() {
1878 if (isMultiUnevenColumnFamilies(getConf())) {
1879 return Sets.newHashSet(Bytes.toString(FAMILY_NAME), Bytes.toString(BIG_FAMILY_NAME),
1880 Bytes.toString(TINY_FAMILY_NAME));
1881 } else {
1882 return Sets.newHashSet(Bytes.toString(FAMILY_NAME));
1883 }
1884 }
1885
1886 private static void setJobConf(Job job, int numMappers, long numNodes,
1887 Integer width, Integer wrapMultiplier, Integer numWalkers) {
1888 job.getConfiguration().setInt(GENERATOR_NUM_MAPPERS_KEY, numMappers);
1889 job.getConfiguration().setLong(GENERATOR_NUM_ROWS_PER_MAP_KEY, numNodes);
1890 if (width != null) {
1891 job.getConfiguration().setInt(GENERATOR_WIDTH_KEY, width);
1892 }
1893 if (wrapMultiplier != null) {
1894 job.getConfiguration().setInt(GENERATOR_WRAP_KEY, wrapMultiplier);
1895 }
1896 if (numWalkers != null) {
1897 job.getConfiguration().setInt(CONCURRENT_WALKER_KEY, numWalkers);
1898 }
1899 }
1900
1901 public static void setJobScannerConf(Job job) {
1902
1903 job.getConfiguration().setBoolean(ScannerCallable.LOG_SCANNER_ACTIVITY, true);
1904 job.getConfiguration().setInt(TableRecordReaderImpl.LOG_PER_ROW_COUNT, 100000);
1905 }
1906
1907 public static void main(String[] args) throws Exception {
1908 Configuration conf = HBaseConfiguration.create();
1909 IntegrationTestingUtility.setUseDistributedCluster(conf);
1910 int ret = ToolRunner.run(conf, new IntegrationTestBigLinkedList(), args);
1911 System.exit(ret);
1912 }
1913 }