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 if (counters == null) {
819 LOG.info("Counters object was null, Generator verification cannot be performed."
820 + " This is commonly a result of insufficient YARN configuration.");
821 return false;
822 }
823
824 if (counters.findCounter(Counts.TERMINATING).getValue() > 0 ||
825 counters.findCounter(Counts.UNDEFINED).getValue() > 0 ||
826 counters.findCounter(Counts.IOEXCEPTION).getValue() > 0) {
827 LOG.error("Concurrent walker failed to verify during Generation phase");
828 LOG.error("TERMINATING nodes: " + counters.findCounter(Counts.TERMINATING).getValue());
829 LOG.error("UNDEFINED nodes: " + counters.findCounter(Counts.UNDEFINED).getValue());
830 LOG.error("IOEXCEPTION nodes: " + counters.findCounter(Counts.IOEXCEPTION).getValue());
831 return false;
832 }
833 } catch (IOException e) {
834 LOG.info("Generator verification could not find counter");
835 return false;
836 }
837 return true;
838 }
839 }
840
841
842
843
844
845
846
847 static class Search extends Configured implements Tool {
848 private static final Log LOG = LogFactory.getLog(Search.class);
849 protected Job job;
850
851 private static void printUsage(final String error) {
852 if (error != null && error.length() > 0) System.out.println("ERROR: " + error);
853 System.err.println("Usage: search <KEYS_DIR> [<MAPPERS_COUNT>]");
854 }
855
856 @Override
857 public int run(String[] args) throws Exception {
858 if (args.length < 1 || args.length > 2) {
859 printUsage(null);
860 return 1;
861 }
862 Path inputDir = new Path(args[0]);
863 int numMappers = 1;
864 if (args.length > 1) {
865 numMappers = Integer.parseInt(args[1]);
866 }
867 return run(inputDir, numMappers);
868 }
869
870
871
872
873 public static class WALSearcher extends WALPlayer {
874 public WALSearcher(Configuration conf) {
875 super(conf);
876 }
877
878
879
880
881 public static class WALMapperSearcher extends WALMapper {
882 private SortedSet<byte []> keysToFind;
883 private AtomicInteger rows = new AtomicInteger(0);
884
885 @Override
886 public void setup(Mapper<WALKey, WALEdit, ImmutableBytesWritable, Mutation>.Context context)
887 throws IOException {
888 super.setup(context);
889 try {
890 this.keysToFind = readKeysToSearch(context.getConfiguration());
891 LOG.info("Loaded keys to find: count=" + this.keysToFind.size());
892 } catch (InterruptedException e) {
893 throw new InterruptedIOException(e.toString());
894 }
895 }
896
897 @Override
898 protected boolean filter(Context context, Cell cell) {
899
900 byte [] row = new byte [cell.getRowLength()];
901 System.arraycopy(cell.getRowArray(), cell.getRowOffset(), row, 0, cell.getRowLength());
902 boolean b = this.keysToFind.contains(row);
903 if (b) {
904 String keyStr = Bytes.toStringBinary(row);
905 try {
906 LOG.info("Found cell=" + cell + " , walKey=" + context.getCurrentKey());
907 } catch (IOException|InterruptedException e) {
908 LOG.warn(e);
909 }
910 if (rows.addAndGet(1) < MISSING_ROWS_TO_LOG) {
911 context.getCounter(FOUND_GROUP_KEY, keyStr).increment(1);
912 }
913 context.getCounter(FOUND_GROUP_KEY, "CELL_WITH_MISSING_ROW").increment(1);
914 }
915 return b;
916 }
917 }
918
919
920 @Override
921 public Job createSubmittableJob(String[] args) throws IOException {
922 Job job = super.createSubmittableJob(args);
923
924 job.setJarByClass(WALMapperSearcher.class);
925 job.setMapperClass(WALMapperSearcher.class);
926 job.setOutputFormatClass(NullOutputFormat.class);
927 return job;
928 }
929 }
930
931 static final String FOUND_GROUP_KEY = "Found";
932 static final String SEARCHER_INPUTDIR_KEY = "searcher.keys.inputdir";
933
934 public int run(Path inputDir, int numMappers) throws Exception {
935 getConf().set(SEARCHER_INPUTDIR_KEY, inputDir.toString());
936 SortedSet<byte []> keys = readKeysToSearch(getConf());
937 if (keys.isEmpty()) throw new RuntimeException("No keys to find");
938 LOG.info("Count of keys to find: " + keys.size());
939 for(byte [] key: keys) LOG.info("Key: " + Bytes.toStringBinary(key));
940 Path hbaseDir = new Path(getConf().get(HConstants.HBASE_DIR));
941
942 Path walsDir = new Path(hbaseDir, HConstants.HREGION_LOGDIR_NAME);
943 Path oldWalsDir = new Path(hbaseDir, HConstants.HREGION_OLDLOGDIR_NAME);
944 LOG.info("Running Search with keys inputDir=" + inputDir +", numMappers=" + numMappers +
945 " against " + getConf().get(HConstants.HBASE_DIR));
946 int ret = ToolRunner.run(new WALSearcher(getConf()), new String [] {walsDir.toString(), ""});
947 if (ret != 0) return ret;
948 return ToolRunner.run(new WALSearcher(getConf()), new String [] {oldWalsDir.toString(), ""});
949 }
950
951 static SortedSet<byte []> readKeysToSearch(final Configuration conf)
952 throws IOException, InterruptedException {
953 Path keysInputDir = new Path(conf.get(SEARCHER_INPUTDIR_KEY));
954 FileSystem fs = FileSystem.get(conf);
955 SortedSet<byte []> result = new TreeSet<byte []>(Bytes.BYTES_COMPARATOR);
956 if (!fs.exists(keysInputDir)) {
957 throw new FileNotFoundException(keysInputDir.toString());
958 }
959 if (!fs.isDirectory(keysInputDir)) {
960 throw new UnsupportedOperationException("TODO");
961 } else {
962 RemoteIterator<LocatedFileStatus> iterator = fs.listFiles(keysInputDir, false);
963 while(iterator.hasNext()) {
964 LocatedFileStatus keyFileStatus = iterator.next();
965
966 if (keyFileStatus.getPath().getName().startsWith("_")) continue;
967 result.addAll(readFileToSearch(conf, fs, keyFileStatus));
968 }
969 }
970 return result;
971 }
972
973 private static SortedSet<byte[]> readFileToSearch(final Configuration conf,
974 final FileSystem fs, final LocatedFileStatus keyFileStatus) throws IOException,
975 InterruptedException {
976 SortedSet<byte []> result = new TreeSet<byte []>(Bytes.BYTES_COMPARATOR);
977
978
979 TaskAttemptContext context = new TaskAttemptContextImpl(conf, new TaskAttemptID());
980 try (SequenceFileAsBinaryInputFormat.SequenceFileAsBinaryRecordReader rr =
981 new SequenceFileAsBinaryInputFormat.SequenceFileAsBinaryRecordReader()) {
982 InputSplit is =
983 new FileSplit(keyFileStatus.getPath(), 0, keyFileStatus.getLen(), new String [] {});
984 rr.initialize(is, context);
985 while (rr.nextKeyValue()) {
986 rr.getCurrentKey();
987 BytesWritable bw = rr.getCurrentValue();
988 if (Verify.VerifyReducer.whichType(bw.getBytes()) == Verify.Counts.UNDEFINED) {
989 byte[] key = new byte[rr.getCurrentKey().getLength()];
990 System.arraycopy(rr.getCurrentKey().getBytes(), 0, key, 0, rr.getCurrentKey()
991 .getLength());
992 result.add(key);
993 }
994 }
995 }
996 return result;
997 }
998 }
999
1000
1001
1002
1003
1004 static class Verify extends Configured implements Tool {
1005
1006 private static final Log LOG = LogFactory.getLog(Verify.class);
1007 protected static final BytesWritable DEF = new BytesWritable(new byte[] { 0 });
1008 protected static final BytesWritable DEF_LOST_FAMILIES = new BytesWritable(new byte[] { 1 });
1009
1010 protected Job job;
1011
1012 public static class VerifyMapper extends TableMapper<BytesWritable, BytesWritable> {
1013 private BytesWritable row = new BytesWritable();
1014 private BytesWritable ref = new BytesWritable();
1015
1016 private boolean multipleUnevenColumnFamilies;
1017
1018 @Override
1019 protected void setup(
1020 Mapper<ImmutableBytesWritable, Result, BytesWritable, BytesWritable>.Context context)
1021 throws IOException, InterruptedException {
1022 this.multipleUnevenColumnFamilies = isMultiUnevenColumnFamilies(context.getConfiguration());
1023 }
1024
1025 @Override
1026 protected void map(ImmutableBytesWritable key, Result value, Context context)
1027 throws IOException ,InterruptedException {
1028 byte[] rowKey = key.get();
1029 row.set(rowKey, 0, rowKey.length);
1030 if (multipleUnevenColumnFamilies
1031 && (!value.containsColumn(BIG_FAMILY_NAME, BIG_FAMILY_NAME) || !value.containsColumn(
1032 TINY_FAMILY_NAME, TINY_FAMILY_NAME))) {
1033 context.write(row, DEF_LOST_FAMILIES);
1034 } else {
1035 context.write(row, DEF);
1036 }
1037 byte[] prev = value.getValue(FAMILY_NAME, COLUMN_PREV);
1038 if (prev != null && prev.length > 0) {
1039 ref.set(prev, 0, prev.length);
1040 context.write(ref, row);
1041 } else {
1042 LOG.warn(String.format("Prev is not set for: %s", Bytes.toStringBinary(rowKey)));
1043 }
1044 }
1045 }
1046
1047
1048
1049
1050
1051 public static enum Counts {
1052 UNREFERENCED, UNDEFINED, REFERENCED, CORRUPT, EXTRAREFERENCES, EXTRA_UNDEF_REFERENCES,
1053 LOST_FAMILIES
1054 }
1055
1056
1057
1058
1059
1060
1061 public static class VerifyReducer extends
1062 Reducer<BytesWritable, BytesWritable, BytesWritable, BytesWritable> {
1063 private ArrayList<byte[]> refs = new ArrayList<byte[]>();
1064 private final BytesWritable UNREF = new BytesWritable(addPrefixFlag(
1065 Counts.UNREFERENCED.ordinal(), new byte[] {}));
1066 private final BytesWritable LOSTFAM = new BytesWritable(addPrefixFlag(
1067 Counts.LOST_FAMILIES.ordinal(), new byte[] {}));
1068
1069 private AtomicInteger rows = new AtomicInteger(0);
1070 private Connection connection;
1071
1072 @Override
1073 protected void setup(Reducer<BytesWritable, BytesWritable, BytesWritable, BytesWritable>.Context context)
1074 throws IOException, InterruptedException {
1075 super.setup(context);
1076 this.connection = ConnectionFactory.createConnection(context.getConfiguration());
1077 }
1078
1079 @Override
1080 protected void cleanup(
1081 Reducer<BytesWritable, BytesWritable, BytesWritable, BytesWritable>.Context context)
1082 throws IOException, InterruptedException {
1083 if (this.connection != null) {
1084 this.connection.close();
1085 }
1086 super.cleanup(context);
1087 }
1088
1089
1090
1091
1092
1093
1094
1095 public static byte[] addPrefixFlag(final int ordinal, final byte [] r) {
1096 byte[] prefix = Bytes.toBytes((short)ordinal);
1097 if (prefix.length != Bytes.SIZEOF_SHORT) {
1098 throw new RuntimeException("Unexpected size: " + prefix.length);
1099 }
1100 byte[] result = new byte[prefix.length + r.length];
1101 System.arraycopy(prefix, 0, result, 0, prefix.length);
1102 System.arraycopy(r, 0, result, prefix.length, r.length);
1103 return result;
1104 }
1105
1106
1107
1108
1109
1110
1111 public static Counts whichType(final byte [] bs) {
1112 int ordinal = Bytes.toShort(bs, 0, Bytes.SIZEOF_SHORT);
1113 return Counts.values()[ordinal];
1114 }
1115
1116
1117
1118
1119
1120 public static byte[] getRowOnly(BytesWritable bw) {
1121 byte[] bytes = new byte [bw.getLength() - Bytes.SIZEOF_SHORT];
1122 System.arraycopy(bw.getBytes(), Bytes.SIZEOF_SHORT, bytes, 0, bytes.length);
1123 return bytes;
1124 }
1125
1126 @Override
1127 public void reduce(BytesWritable key, Iterable<BytesWritable> values, Context context)
1128 throws IOException, InterruptedException {
1129 int defCount = 0;
1130 boolean lostFamilies = false;
1131 refs.clear();
1132 for (BytesWritable type : values) {
1133 if (type.getLength() == DEF.getLength()) {
1134 defCount++;
1135 if (type.getBytes()[0] == 1) {
1136 lostFamilies = true;
1137 }
1138 } else {
1139 byte[] bytes = new byte[type.getLength()];
1140 System.arraycopy(type.getBytes(), 0, bytes, 0, type.getLength());
1141 refs.add(bytes);
1142 }
1143 }
1144
1145
1146 StringBuilder refsSb = null;
1147 String keyString = Bytes.toStringBinary(key.getBytes(), 0, key.getLength());
1148 if (defCount == 0 || refs.size() != 1) {
1149 refsSb = dumpExtraInfoOnRefs(key, context, refs);
1150 LOG.error("LinkedListError: key=" + keyString + ", reference(s)=" +
1151 (refsSb != null? refsSb.toString(): ""));
1152 }
1153 if (lostFamilies) {
1154 LOG.error("LinkedListError: key=" + keyString + ", lost big or tiny families");
1155 context.getCounter(Counts.LOST_FAMILIES).increment(1);
1156 context.write(key, LOSTFAM);
1157 }
1158
1159 if (defCount == 0 && refs.size() > 0) {
1160
1161
1162
1163 for (int i = 0; i < refs.size(); i++) {
1164 byte[] bs = refs.get(i);
1165 int ordinal;
1166 if (i <= 0) {
1167 ordinal = Counts.UNDEFINED.ordinal();
1168 context.write(key, new BytesWritable(addPrefixFlag(ordinal, bs)));
1169 context.getCounter(Counts.UNDEFINED).increment(1);
1170 } else {
1171 ordinal = Counts.EXTRA_UNDEF_REFERENCES.ordinal();
1172 context.write(key, new BytesWritable(addPrefixFlag(ordinal, bs)));
1173 }
1174 }
1175 if (rows.addAndGet(1) < MISSING_ROWS_TO_LOG) {
1176
1177
1178
1179
1180 context.getCounter("undef", keyString).increment(1);
1181 }
1182 } else if (defCount > 0 && refs.size() == 0) {
1183
1184 context.write(key, UNREF);
1185 context.getCounter(Counts.UNREFERENCED).increment(1);
1186 if (rows.addAndGet(1) < MISSING_ROWS_TO_LOG) {
1187 context.getCounter("unref", keyString).increment(1);
1188 }
1189 } else {
1190 if (refs.size() > 1) {
1191
1192 for (int i = 1; i < refs.size(); i++) {
1193 context.write(key,
1194 new BytesWritable(addPrefixFlag(Counts.EXTRAREFERENCES.ordinal(), refs.get(i))));
1195 }
1196 context.getCounter(Counts.EXTRAREFERENCES).increment(refs.size() - 1);
1197 }
1198
1199 context.getCounter(Counts.REFERENCED).increment(1);
1200 }
1201 }
1202
1203
1204
1205
1206
1207
1208 private StringBuilder dumpExtraInfoOnRefs(final BytesWritable key, final Context context,
1209 final List<byte []> refs)
1210 throws IOException {
1211 StringBuilder refsSb = null;
1212 if (refs.isEmpty()) return refsSb;
1213 refsSb = new StringBuilder();
1214 String comma = "";
1215
1216
1217
1218 TableName tn = getTableName(context.getConfiguration());
1219 try (Table t = this.connection.getTable(tn)) {
1220 for (byte [] ref : refs) {
1221 Result r = t.get(new Get(ref));
1222 List<Cell> cells = r.listCells();
1223 String ts = (cells != null && !cells.isEmpty())?
1224 new java.util.Date(cells.get(0).getTimestamp()).toString(): "";
1225 byte [] b = r.getValue(FAMILY_NAME, COLUMN_CLIENT);
1226 String jobStr = (b != null && b.length > 0)? Bytes.toString(b): "";
1227 b = r.getValue(FAMILY_NAME, COLUMN_COUNT);
1228 long count = (b != null && b.length > 0)? Bytes.toLong(b): -1;
1229 b = r.getValue(FAMILY_NAME, COLUMN_PREV);
1230 String refRegionLocation = "";
1231 String keyRegionLocation = "";
1232 if (b != null && b.length > 0) {
1233 try (RegionLocator rl = this.connection.getRegionLocator(tn)) {
1234 HRegionLocation hrl = rl.getRegionLocation(b);
1235 if (hrl != null) refRegionLocation = hrl.toString();
1236
1237 hrl = rl.getRegionLocation(key.getBytes());
1238 if (hrl != null) keyRegionLocation = hrl.toString();
1239 }
1240 }
1241 LOG.error("Extras on ref without a def, ref=" + Bytes.toStringBinary(ref) +
1242 ", refPrevEqualsKey=" +
1243 (Bytes.compareTo(key.getBytes(), 0, key.getLength(), b, 0, b.length) == 0) +
1244 ", key=" + Bytes.toStringBinary(key.getBytes(), 0, key.getLength()) +
1245 ", ref row date=" + ts + ", jobStr=" + jobStr +
1246 ", ref row count=" + count +
1247 ", ref row regionLocation=" + refRegionLocation +
1248 ", key row regionLocation=" + keyRegionLocation);
1249 refsSb.append(comma);
1250 comma = ",";
1251 refsSb.append(Bytes.toStringBinary(ref));
1252 }
1253 }
1254 return refsSb;
1255 }
1256 }
1257
1258 @Override
1259 public int run(String[] args) throws Exception {
1260 if (args.length != 2) {
1261 System.out.println("Usage : " + Verify.class.getSimpleName()
1262 + " <output dir> <num reducers>");
1263 return 0;
1264 }
1265
1266 String outputDir = args[0];
1267 int numReducers = Integer.parseInt(args[1]);
1268
1269 return run(outputDir, numReducers);
1270 }
1271
1272 public int run(String outputDir, int numReducers) throws Exception {
1273 return run(new Path(outputDir), numReducers);
1274 }
1275
1276 public int run(Path outputDir, int numReducers) throws Exception {
1277 LOG.info("Running Verify with outputDir=" + outputDir +", numReducers=" + numReducers);
1278
1279 job = Job.getInstance(getConf());
1280
1281 job.setJobName("Link Verifier");
1282 job.setNumReduceTasks(numReducers);
1283 job.setJarByClass(getClass());
1284
1285 setJobScannerConf(job);
1286
1287 Scan scan = new Scan();
1288 scan.addColumn(FAMILY_NAME, COLUMN_PREV);
1289 scan.setCaching(10000);
1290 scan.setCacheBlocks(false);
1291 if (isMultiUnevenColumnFamilies(getConf())) {
1292 scan.addColumn(BIG_FAMILY_NAME, BIG_FAMILY_NAME);
1293 scan.addColumn(TINY_FAMILY_NAME, TINY_FAMILY_NAME);
1294 }
1295
1296 TableMapReduceUtil.initTableMapperJob(getTableName(getConf()).getName(), scan,
1297 VerifyMapper.class, BytesWritable.class, BytesWritable.class, job);
1298 TableMapReduceUtil.addDependencyJarsForClasses(job.getConfiguration(),
1299 AbstractHBaseTool.class);
1300
1301 job.getConfiguration().setBoolean("mapreduce.map.speculative", false);
1302
1303 job.setReducerClass(VerifyReducer.class);
1304 job.setOutputFormatClass(SequenceFileAsBinaryOutputFormat.class);
1305 job.setOutputKeyClass(BytesWritable.class);
1306 job.setOutputValueClass(BytesWritable.class);
1307 TextOutputFormat.setOutputPath(job, outputDir);
1308
1309 boolean success = job.waitForCompletion(true);
1310
1311 if (success) {
1312 Counters counters = job.getCounters();
1313 if (null == counters) {
1314 LOG.warn("Counters were null, cannot verify Job completion."
1315 + " This is commonly a result of insufficient YARN configuration.");
1316
1317 return 0;
1318 }
1319
1320
1321 if (verifyUnexpectedValues(counters)) {
1322
1323 return 0;
1324 }
1325 }
1326
1327
1328 return 1;
1329 }
1330
1331 public boolean verify(long expectedReferenced) throws Exception {
1332 if (job == null) {
1333 throw new IllegalStateException("You should call run() first");
1334 }
1335
1336 Counters counters = job.getCounters();
1337 if (counters == null) {
1338 LOG.info("Counters object was null, write verification cannot be performed."
1339 + " This is commonly a result of insufficient YARN configuration.");
1340 return false;
1341 }
1342
1343
1344 boolean success = verifyExpectedValues(expectedReferenced, counters);
1345
1346 if (!verifyUnexpectedValues(counters)) {
1347
1348 success = false;
1349 }
1350
1351 if (!success) {
1352 handleFailure(counters);
1353 }
1354 return success;
1355 }
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366 protected boolean verifyExpectedValues(long expectedReferenced, Counters counters) {
1367 final Counter referenced = counters.findCounter(Counts.REFERENCED);
1368 final Counter unreferenced = counters.findCounter(Counts.UNREFERENCED);
1369 boolean success = true;
1370
1371 if (expectedReferenced != referenced.getValue()) {
1372 LOG.error("Expected referenced count does not match with actual referenced count. " +
1373 "expected referenced=" + expectedReferenced + " ,actual=" + referenced.getValue());
1374 success = false;
1375 }
1376
1377 if (unreferenced.getValue() > 0) {
1378 final Counter multiref = counters.findCounter(Counts.EXTRAREFERENCES);
1379 boolean couldBeMultiRef = (multiref.getValue() == unreferenced.getValue());
1380 LOG.error("Unreferenced nodes were not expected. Unreferenced count=" + unreferenced.getValue()
1381 + (couldBeMultiRef ? "; could be due to duplicate random numbers" : ""));
1382 success = false;
1383 }
1384
1385 return success;
1386 }
1387
1388
1389
1390
1391
1392
1393
1394
1395 protected boolean verifyUnexpectedValues(Counters counters) {
1396 final Counter undefined = counters.findCounter(Counts.UNDEFINED);
1397 final Counter lostfamilies = counters.findCounter(Counts.LOST_FAMILIES);
1398 boolean success = true;
1399
1400 if (undefined.getValue() > 0) {
1401 LOG.error("Found an undefined node. Undefined count=" + undefined.getValue());
1402 success = false;
1403 }
1404
1405 if (lostfamilies.getValue() > 0) {
1406 LOG.error("Found nodes which lost big or tiny families, count=" + lostfamilies.getValue());
1407 success = false;
1408 }
1409
1410 return success;
1411 }
1412
1413 protected void handleFailure(Counters counters) throws IOException {
1414 Configuration conf = job.getConfiguration();
1415 TableName tableName = getTableName(conf);
1416 try (Connection conn = ConnectionFactory.createConnection(conf)) {
1417 try (RegionLocator rl = conn.getRegionLocator(tableName)) {
1418 CounterGroup g = counters.getGroup("undef");
1419 Iterator<Counter> it = g.iterator();
1420 while (it.hasNext()) {
1421 String keyString = it.next().getName();
1422 byte[] key = Bytes.toBytes(keyString);
1423 HRegionLocation loc = rl.getRegionLocation(key, true);
1424 LOG.error("undefined row " + keyString + ", " + loc);
1425 }
1426 g = counters.getGroup("unref");
1427 it = g.iterator();
1428 while (it.hasNext()) {
1429 String keyString = it.next().getName();
1430 byte[] key = Bytes.toBytes(keyString);
1431 HRegionLocation loc = rl.getRegionLocation(key, true);
1432 LOG.error("unreferred row " + keyString + ", " + loc);
1433 }
1434 }
1435 }
1436 }
1437 }
1438
1439
1440
1441
1442
1443 static class Loop extends Configured implements Tool {
1444
1445 private static final Log LOG = LogFactory.getLog(Loop.class);
1446 private static final String USAGE = "Usage: Loop <num iterations> <num mappers> " +
1447 "<num nodes per mapper> <output dir> <num reducers> [<width> <wrap multiplier>" +
1448 " <num walker threads>] \n" +
1449 "where <num nodes per map> should be a multiple of width*wrap multiplier, 25M by default \n" +
1450 "walkers will select and verify random flushed loop during Generation.";
1451
1452 IntegrationTestBigLinkedList it;
1453
1454 protected void runGenerator(int numMappers, long numNodes,
1455 String outputDir, Integer width, Integer wrapMultiplier, Integer numWalkers)
1456 throws Exception {
1457 Path outputPath = new Path(outputDir);
1458 UUID uuid = UUID.randomUUID();
1459 Path generatorOutput = new Path(outputPath, uuid.toString());
1460
1461 Generator generator = new Generator();
1462 generator.setConf(getConf());
1463 int retCode = generator.run(numMappers, numNodes, generatorOutput, width, wrapMultiplier,
1464 numWalkers);
1465 if (retCode > 0) {
1466 throw new RuntimeException("Generator failed with return code: " + retCode);
1467 }
1468 if (numWalkers > 0) {
1469 if (!generator.verify()) {
1470 throw new RuntimeException("Generator.verify failed");
1471 }
1472 }
1473 }
1474
1475 protected void runVerify(String outputDir,
1476 int numReducers, long expectedNumNodes) throws Exception {
1477 Path outputPath = new Path(outputDir);
1478 UUID uuid = UUID.randomUUID();
1479 Path iterationOutput = new Path(outputPath, uuid.toString());
1480
1481 Verify verify = new Verify();
1482 verify.setConf(getConf());
1483 int retCode = verify.run(iterationOutput, numReducers);
1484 if (retCode > 0) {
1485 throw new RuntimeException("Verify.run failed with return code: " + retCode);
1486 }
1487
1488 if (!verify.verify(expectedNumNodes)) {
1489 throw new RuntimeException("Verify.verify failed");
1490 }
1491 LOG.info("Verify finished with success. Total nodes=" + expectedNumNodes);
1492 }
1493
1494 @Override
1495 public int run(String[] args) throws Exception {
1496 if (args.length < 5) {
1497 System.err.println(USAGE);
1498 return 1;
1499 }
1500 try {
1501 int numIterations = Integer.parseInt(args[0]);
1502 int numMappers = Integer.parseInt(args[1]);
1503 long numNodes = Long.parseLong(args[2]);
1504 String outputDir = args[3];
1505 int numReducers = Integer.parseInt(args[4]);
1506 Integer width = (args.length < 6) ? null : Integer.parseInt(args[5]);
1507 Integer wrapMultiplier = (args.length < 7) ? null : Integer.parseInt(args[6]);
1508 Integer numWalkers = (args.length < 8) ? 0 : Integer.parseInt(args[7]);
1509
1510 long expectedNumNodes = 0;
1511
1512 if (numIterations < 0) {
1513 numIterations = Integer.MAX_VALUE;
1514 }
1515 LOG.info("Running Loop with args:" + Arrays.deepToString(args));
1516 for (int i = 0; i < numIterations; i++) {
1517 LOG.info("Starting iteration = " + i);
1518 runGenerator(numMappers, numNodes, outputDir, width, wrapMultiplier, numWalkers);
1519 expectedNumNodes += numMappers * numNodes;
1520 runVerify(outputDir, numReducers, expectedNumNodes);
1521 }
1522 return 0;
1523 } catch (NumberFormatException e) {
1524 System.err.println("Parsing loop arguments failed: " + e.getMessage());
1525 System.err.println(USAGE);
1526 return 1;
1527 }
1528 }
1529 }
1530
1531
1532
1533
1534 private static class Print extends Configured implements Tool {
1535 @Override
1536 public int run(String[] args) throws Exception {
1537 Options options = new Options();
1538 options.addOption("s", "start", true, "start key");
1539 options.addOption("e", "end", true, "end key");
1540 options.addOption("l", "limit", true, "number to print");
1541
1542 GnuParser parser = new GnuParser();
1543 CommandLine cmd = null;
1544 try {
1545 cmd = parser.parse(options, args);
1546 if (cmd.getArgs().length != 0) {
1547 throw new ParseException("Command takes no arguments");
1548 }
1549 } catch (ParseException e) {
1550 System.err.println("Failed to parse command line " + e.getMessage());
1551 System.err.println();
1552 HelpFormatter formatter = new HelpFormatter();
1553 formatter.printHelp(getClass().getSimpleName(), options);
1554 System.exit(-1);
1555 }
1556
1557 Table table = new HTable(getConf(), getTableName(getConf()));
1558
1559 Scan scan = new Scan();
1560 scan.setBatch(10000);
1561
1562 if (cmd.hasOption("s"))
1563 scan.setStartRow(Bytes.toBytesBinary(cmd.getOptionValue("s")));
1564
1565 if (cmd.hasOption("e"))
1566 scan.setStopRow(Bytes.toBytesBinary(cmd.getOptionValue("e")));
1567
1568 int limit = 0;
1569 if (cmd.hasOption("l"))
1570 limit = Integer.parseInt(cmd.getOptionValue("l"));
1571 else
1572 limit = 100;
1573
1574 ResultScanner scanner = table.getScanner(scan);
1575
1576 CINode node = new CINode();
1577 Result result = scanner.next();
1578 int count = 0;
1579 while (result != null && count++ < limit) {
1580 node = getCINode(result, node);
1581 System.out.printf("%s:%s:%012d:%s\n", Bytes.toStringBinary(node.key),
1582 Bytes.toStringBinary(node.prev), node.count, node.client);
1583 result = scanner.next();
1584 }
1585 scanner.close();
1586 table.close();
1587
1588 return 0;
1589 }
1590 }
1591
1592
1593
1594
1595 private static class Delete extends Configured implements Tool {
1596 @Override
1597 public int run(String[] args) throws Exception {
1598 if (args.length != 1) {
1599 System.out.println("Usage : " + Delete.class.getSimpleName() + " <node to delete>");
1600 return 0;
1601 }
1602 byte[] val = Bytes.toBytesBinary(args[0]);
1603
1604 org.apache.hadoop.hbase.client.Delete delete
1605 = new org.apache.hadoop.hbase.client.Delete(val);
1606
1607 Table table = new HTable(getConf(), getTableName(getConf()));
1608 table.delete(delete);
1609 table.close();
1610
1611 System.out.println("Delete successful");
1612 return 0;
1613 }
1614 }
1615
1616 abstract static class WalkerBase extends Configured{
1617 protected static CINode findStartNode(Table table, byte[] startKey) throws IOException {
1618 Scan scan = new Scan();
1619 scan.setStartRow(startKey);
1620 scan.setBatch(1);
1621 scan.addColumn(FAMILY_NAME, COLUMN_PREV);
1622
1623 long t1 = System.currentTimeMillis();
1624 ResultScanner scanner = table.getScanner(scan);
1625 Result result = scanner.next();
1626 long t2 = System.currentTimeMillis();
1627 scanner.close();
1628
1629 if ( result != null) {
1630 CINode node = getCINode(result, new CINode());
1631 System.out.printf("FSR %d %s\n", t2 - t1, Bytes.toStringBinary(node.key));
1632 return node;
1633 }
1634
1635 System.out.println("FSR " + (t2 - t1));
1636
1637 return null;
1638 }
1639 protected CINode getNode(byte[] row, Table table, CINode node) throws IOException {
1640 Get get = new Get(row);
1641 get.addColumn(FAMILY_NAME, COLUMN_PREV);
1642 Result result = table.get(get);
1643 return getCINode(result, node);
1644 }
1645 }
1646
1647
1648
1649
1650 private static class Walker extends WalkerBase implements Tool {
1651
1652 public Walker(){}
1653
1654 @Override
1655 public int run(String[] args) throws IOException {
1656
1657 Options options = new Options();
1658 options.addOption("n", "num", true, "number of queries");
1659 options.addOption("s", "start", true, "key to start at, binary string");
1660 options.addOption("l", "logevery", true, "log every N queries");
1661
1662 GnuParser parser = new GnuParser();
1663 CommandLine cmd = null;
1664 try {
1665 cmd = parser.parse(options, args);
1666 if (cmd.getArgs().length != 0) {
1667 throw new ParseException("Command takes no arguments");
1668 }
1669 } catch (ParseException e) {
1670 System.err.println("Failed to parse command line " + e.getMessage());
1671 System.err.println();
1672 HelpFormatter formatter = new HelpFormatter();
1673 formatter.printHelp(getClass().getSimpleName(), options);
1674 System.exit(-1);
1675 }
1676
1677 long maxQueries = Long.MAX_VALUE;
1678 if (cmd.hasOption('n')) {
1679 maxQueries = Long.parseLong(cmd.getOptionValue("n"));
1680 }
1681 Random rand = new SecureRandom();
1682 boolean isSpecificStart = cmd.hasOption('s');
1683
1684 byte[] startKey = isSpecificStart ? Bytes.toBytesBinary(cmd.getOptionValue('s')) : null;
1685 int logEvery = cmd.hasOption('l') ? Integer.parseInt(cmd.getOptionValue('l')) : 1;
1686
1687 Table table = new HTable(getConf(), getTableName(getConf()));
1688 long numQueries = 0;
1689
1690
1691
1692 while (numQueries < maxQueries && (numQueries == 0 || !isSpecificStart)) {
1693 if (!isSpecificStart) {
1694 startKey = new byte[ROWKEY_LENGTH];
1695 rand.nextBytes(startKey);
1696 }
1697 CINode node = findStartNode(table, startKey);
1698 if (node == null && isSpecificStart) {
1699 System.err.printf("Start node not found: %s \n", Bytes.toStringBinary(startKey));
1700 }
1701 numQueries++;
1702 while (node != null && node.prev.length != NO_KEY.length &&
1703 numQueries < maxQueries) {
1704 byte[] prev = node.prev;
1705 long t1 = System.currentTimeMillis();
1706 node = getNode(prev, table, node);
1707 long t2 = System.currentTimeMillis();
1708 if (logEvery > 0 && numQueries % logEvery == 0) {
1709 System.out.printf("CQ %d: %d %s \n", numQueries, t2 - t1, Bytes.toStringBinary(prev));
1710 }
1711 numQueries++;
1712 if (node == null) {
1713 System.err.printf("UNDEFINED NODE %s \n", Bytes.toStringBinary(prev));
1714 } else if (node.prev.length == NO_KEY.length) {
1715 System.err.printf("TERMINATING NODE %s \n", Bytes.toStringBinary(node.key));
1716 }
1717 }
1718 }
1719 table.close();
1720 return 0;
1721 }
1722 }
1723
1724 private static class Clean extends Configured implements Tool {
1725 @Override public int run(String[] args) throws Exception {
1726 if (args.length < 1) {
1727 System.err.println("Usage: Clean <output dir>");
1728 return -1;
1729 }
1730
1731 Path p = new Path(args[0]);
1732 Configuration conf = getConf();
1733 TableName tableName = getTableName(conf);
1734 try (FileSystem fs = HFileSystem.get(conf);
1735 Connection conn = ConnectionFactory.createConnection(conf);
1736 Admin admin = conn.getAdmin()) {
1737 if (admin.tableExists(tableName)) {
1738 admin.disableTable(tableName);
1739 admin.deleteTable(tableName);
1740 }
1741
1742 if (fs.exists(p)) {
1743 fs.delete(p, true);
1744 }
1745 }
1746
1747 return 0;
1748 }
1749 }
1750
1751 static TableName getTableName(Configuration conf) {
1752 return TableName.valueOf(conf.get(TABLE_NAME_KEY, DEFAULT_TABLE_NAME));
1753 }
1754
1755 private static CINode getCINode(Result result, CINode node) {
1756 node.key = Bytes.copy(result.getRow());
1757 if (result.containsColumn(FAMILY_NAME, COLUMN_PREV)) {
1758 node.prev = Bytes.copy(result.getValue(FAMILY_NAME, COLUMN_PREV));
1759 } else {
1760 node.prev = NO_KEY;
1761 }
1762 if (result.containsColumn(FAMILY_NAME, COLUMN_COUNT)) {
1763 node.count = Bytes.toLong(result.getValue(FAMILY_NAME, COLUMN_COUNT));
1764 } else {
1765 node.count = -1;
1766 }
1767 if (result.containsColumn(FAMILY_NAME, COLUMN_CLIENT)) {
1768 node.client = Bytes.toString(result.getValue(FAMILY_NAME, COLUMN_CLIENT));
1769 } else {
1770 node.client = "";
1771 }
1772 return node;
1773 }
1774
1775 protected IntegrationTestingUtility util;
1776
1777 @Override
1778 public void setUpCluster() throws Exception {
1779 util = getTestingUtil(getConf());
1780 boolean isDistributed = util.isDistributedCluster();
1781 util.initializeCluster(isDistributed ? 1 : this.NUM_SLAVES_BASE);
1782 if (!isDistributed) {
1783 util.startMiniMapReduceCluster();
1784 }
1785 this.setConf(util.getConfiguration());
1786 }
1787
1788 @Override
1789 public void cleanUpCluster() throws Exception {
1790 super.cleanUpCluster();
1791 if (util.isDistributedCluster()) {
1792 util.shutdownMiniMapReduceCluster();
1793 }
1794 }
1795
1796 private static boolean isMultiUnevenColumnFamilies(Configuration conf) {
1797 return conf.getBoolean(Generator.MULTIPLE_UNEVEN_COLUMNFAMILIES_KEY,true);
1798 }
1799
1800 @Test
1801 public void testContinuousIngest() throws IOException, Exception {
1802
1803 Configuration conf = getTestingUtil(getConf()).getConfiguration();
1804 if (isMultiUnevenColumnFamilies(getConf())) {
1805
1806 conf.set(FlushPolicyFactory.HBASE_FLUSH_POLICY_KEY, FlushLargeStoresPolicy.class.getName());
1807 }
1808 int ret =
1809 ToolRunner.run(conf, new Loop(), new String[] { "1", "1", "2000000",
1810 util.getDataTestDirOnTestFS("IntegrationTestBigLinkedList").toString(), "1" });
1811 org.junit.Assert.assertEquals(0, ret);
1812 }
1813
1814 private void usage() {
1815 System.err.println("Usage: " + this.getClass().getSimpleName() + " COMMAND [COMMAND options]");
1816 printCommands();
1817 }
1818
1819 private void printCommands() {
1820 System.err.println("Commands:");
1821 System.err.println(" generator Map only job that generates data.");
1822 System.err.println(" verify A map reduce job that looks for holes. Check return code and");
1823 System.err.println(" look at the counts after running. See REFERENCED and");
1824 System.err.println(" UNREFERENCED are ok. Any UNDEFINED counts are bad. Do not run");
1825 System.err.println(" with the Generator.");
1826 System.err.println(" walker " +
1827 "Standalone program that starts following a linked list & emits timing info.");
1828 System.err.println(" print Standalone program that prints nodes in the linked list.");
1829 System.err.println(" delete Standalone program that deletes a·single node.");
1830 System.err.println(" loop Program to Loop through Generator and Verify steps");
1831 System.err.println(" clean Program to clean all left over detritus.");
1832 System.err.println(" search Search for missing keys.");
1833 System.err.flush();
1834 }
1835
1836 @Override
1837 protected void processOptions(CommandLine cmd) {
1838 super.processOptions(cmd);
1839 String[] args = cmd.getArgs();
1840
1841 if (args.length < 1) {
1842 printUsage(this.getClass().getSimpleName() +
1843 " <general options> COMMAND [<COMMAND options>]", "General options:", "");
1844 printCommands();
1845
1846 throw new RuntimeException("Incorrect Number of args.");
1847 }
1848 toRun = args[0];
1849 otherArgs = Arrays.copyOfRange(args, 1, args.length);
1850 }
1851
1852 @Override
1853 public int runTestFromCommandLine() throws Exception {
1854 Tool tool = null;
1855 if (toRun.equalsIgnoreCase("Generator")) {
1856 tool = new Generator();
1857 } else if (toRun.equalsIgnoreCase("Verify")) {
1858 tool = new Verify();
1859 } else if (toRun.equalsIgnoreCase("Loop")) {
1860 Loop loop = new Loop();
1861 loop.it = this;
1862 tool = loop;
1863 } else if (toRun.equalsIgnoreCase("Walker")) {
1864 tool = new Walker();
1865 } else if (toRun.equalsIgnoreCase("Print")) {
1866 tool = new Print();
1867 } else if (toRun.equalsIgnoreCase("Delete")) {
1868 tool = new Delete();
1869 } else if (toRun.equalsIgnoreCase("Clean")) {
1870 tool = new Clean();
1871 } else if (toRun.equalsIgnoreCase("Search")) {
1872 tool = new Search();
1873 } else {
1874 usage();
1875 throw new RuntimeException("Unknown arg");
1876 }
1877
1878 return ToolRunner.run(getConf(), tool, otherArgs);
1879 }
1880
1881 @Override
1882 public TableName getTablename() {
1883 Configuration c = getConf();
1884 return TableName.valueOf(c.get(TABLE_NAME_KEY, DEFAULT_TABLE_NAME));
1885 }
1886
1887 @Override
1888 protected Set<String> getColumnFamilies() {
1889 if (isMultiUnevenColumnFamilies(getConf())) {
1890 return Sets.newHashSet(Bytes.toString(FAMILY_NAME), Bytes.toString(BIG_FAMILY_NAME),
1891 Bytes.toString(TINY_FAMILY_NAME));
1892 } else {
1893 return Sets.newHashSet(Bytes.toString(FAMILY_NAME));
1894 }
1895 }
1896
1897 private static void setJobConf(Job job, int numMappers, long numNodes,
1898 Integer width, Integer wrapMultiplier, Integer numWalkers) {
1899 job.getConfiguration().setInt(GENERATOR_NUM_MAPPERS_KEY, numMappers);
1900 job.getConfiguration().setLong(GENERATOR_NUM_ROWS_PER_MAP_KEY, numNodes);
1901 if (width != null) {
1902 job.getConfiguration().setInt(GENERATOR_WIDTH_KEY, width);
1903 }
1904 if (wrapMultiplier != null) {
1905 job.getConfiguration().setInt(GENERATOR_WRAP_KEY, wrapMultiplier);
1906 }
1907 if (numWalkers != null) {
1908 job.getConfiguration().setInt(CONCURRENT_WALKER_KEY, numWalkers);
1909 }
1910 }
1911
1912 public static void setJobScannerConf(Job job) {
1913
1914 job.getConfiguration().setBoolean(ScannerCallable.LOG_SCANNER_ACTIVITY, true);
1915 job.getConfiguration().setInt(TableRecordReaderImpl.LOG_PER_ROW_COUNT, 100000);
1916 }
1917
1918 public static void main(String[] args) throws Exception {
1919 Configuration conf = HBaseConfiguration.create();
1920 IntegrationTestingUtility.setUseDistributedCluster(conf);
1921 int ret = ToolRunner.run(conf, new IntegrationTestBigLinkedList(), args);
1922 System.exit(ret);
1923 }
1924 }