1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase;
20
21 import static org.codehaus.jackson.map.SerializationConfig.Feature.SORT_PROPERTIES_ALPHABETICALLY;
22
23 import java.io.IOException;
24 import java.io.PrintStream;
25 import java.lang.reflect.Constructor;
26 import java.math.BigDecimal;
27 import java.math.MathContext;
28 import java.text.DecimalFormat;
29 import java.text.SimpleDateFormat;
30 import java.util.ArrayList;
31 import java.util.Arrays;
32 import java.util.Date;
33 import java.util.LinkedList;
34 import java.util.Locale;
35 import java.util.Map;
36 import java.util.Queue;
37 import java.util.Random;
38 import java.util.TreeMap;
39 import java.util.NoSuchElementException;
40 import java.util.concurrent.Callable;
41 import java.util.concurrent.ExecutionException;
42 import java.util.concurrent.ExecutorService;
43 import java.util.concurrent.Executors;
44 import java.util.concurrent.Future;
45
46 import com.google.common.base.Objects;
47 import com.google.common.util.concurrent.ThreadFactoryBuilder;
48
49 import org.apache.commons.lang.StringUtils;
50 import org.apache.commons.logging.Log;
51 import org.apache.commons.logging.LogFactory;
52 import org.apache.hadoop.conf.Configuration;
53 import org.apache.hadoop.conf.Configured;
54 import org.apache.hadoop.fs.FileSystem;
55 import org.apache.hadoop.fs.Path;
56 import org.apache.hadoop.hbase.classification.InterfaceAudience;
57 import org.apache.hadoop.hbase.client.Admin;
58 import org.apache.hadoop.hbase.client.Append;
59 import org.apache.hadoop.hbase.client.BufferedMutator;
60 import org.apache.hadoop.hbase.client.BufferedMutatorParams;
61 import org.apache.hadoop.hbase.client.Connection;
62 import org.apache.hadoop.hbase.client.ConnectionFactory;
63 import org.apache.hadoop.hbase.client.Consistency;
64 import org.apache.hadoop.hbase.client.Delete;
65 import org.apache.hadoop.hbase.client.Durability;
66 import org.apache.hadoop.hbase.client.Get;
67 import org.apache.hadoop.hbase.client.Increment;
68 import org.apache.hadoop.hbase.client.Put;
69 import org.apache.hadoop.hbase.client.Result;
70 import org.apache.hadoop.hbase.client.ResultScanner;
71 import org.apache.hadoop.hbase.client.RowMutations;
72 import org.apache.hadoop.hbase.client.Scan;
73 import org.apache.hadoop.hbase.client.Table;
74 import org.apache.hadoop.hbase.client.metrics.ScanMetrics;
75 import org.apache.hadoop.hbase.filter.BinaryComparator;
76 import org.apache.hadoop.hbase.filter.CompareFilter;
77 import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
78 import org.apache.hadoop.hbase.filter.Filter;
79 import org.apache.hadoop.hbase.filter.FilterAllFilter;
80 import org.apache.hadoop.hbase.filter.FilterList;
81 import org.apache.hadoop.hbase.filter.PageFilter;
82 import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
83 import org.apache.hadoop.hbase.filter.WhileMatchFilter;
84 import org.apache.hadoop.hbase.io.compress.Compression;
85 import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
86 import org.apache.hadoop.hbase.io.hfile.RandomDistribution;
87 import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
88 import org.apache.hadoop.hbase.regionserver.BloomType;
89 import org.apache.hadoop.hbase.trace.HBaseHTraceConfiguration;
90 import org.apache.hadoop.hbase.trace.SpanReceiverHost;
91 import org.apache.hadoop.hbase.util.*;
92 import org.apache.hadoop.io.LongWritable;
93 import org.apache.hadoop.io.Text;
94 import org.apache.hadoop.mapreduce.Job;
95 import org.apache.hadoop.mapreduce.Mapper;
96 import org.apache.hadoop.mapreduce.lib.input.NLineInputFormat;
97 import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
98 import org.apache.hadoop.mapreduce.lib.reduce.LongSumReducer;
99 import org.apache.hadoop.util.Tool;
100 import org.apache.hadoop.util.ToolRunner;
101 import org.codehaus.jackson.map.ObjectMapper;
102
103 import com.yammer.metrics.core.Histogram;
104 import com.yammer.metrics.stats.UniformSample;
105
106 import org.apache.htrace.Sampler;
107 import org.apache.htrace.Trace;
108 import org.apache.htrace.TraceScope;
109 import org.apache.htrace.impl.ProbabilitySampler;
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128 @InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.TOOLS)
129 public class PerformanceEvaluation extends Configured implements Tool {
130 private static final Log LOG = LogFactory.getLog(PerformanceEvaluation.class.getName());
131 private static final ObjectMapper MAPPER = new ObjectMapper();
132 static {
133 MAPPER.configure(SORT_PROPERTIES_ALPHABETICALLY, true);
134 }
135
136 public static final String TABLE_NAME = "TestTable";
137 public static final String FAMILY_NAME_BASE = "info";
138 public static final byte[] FAMILY_ZERO = Bytes.toBytes("info0");
139 public static final byte[] COLUMN_ZERO = Bytes.toBytes("" + 0);
140 public static final int DEFAULT_VALUE_LENGTH = 1000;
141 public static final int ROW_LENGTH = 26;
142
143 private static final int ONE_GB = 1024 * 1024 * 1000;
144 private static final int DEFAULT_ROWS_PER_GB = ONE_GB / DEFAULT_VALUE_LENGTH;
145
146 private static final int TAG_LENGTH = 256;
147 private static final DecimalFormat FMT = new DecimalFormat("0.##");
148 private static final MathContext CXT = MathContext.DECIMAL64;
149 private static final BigDecimal MS_PER_SEC = BigDecimal.valueOf(1000);
150 private static final BigDecimal BYTES_PER_MB = BigDecimal.valueOf(1024 * 1024);
151 private static final TestOptions DEFAULT_OPTS = new TestOptions();
152
153 private static Map<String, CmdDescriptor> COMMANDS = new TreeMap<String, CmdDescriptor>();
154 private static final Path PERF_EVAL_DIR = new Path("performance_evaluation");
155
156 static {
157 addCommandDescriptor(RandomReadTest.class, "randomRead",
158 "Run random read test");
159 addCommandDescriptor(RandomSeekScanTest.class, "randomSeekScan",
160 "Run random seek and scan 100 test");
161 addCommandDescriptor(RandomScanWithRange10Test.class, "scanRange10",
162 "Run random seek scan with both start and stop row (max 10 rows)");
163 addCommandDescriptor(RandomScanWithRange100Test.class, "scanRange100",
164 "Run random seek scan with both start and stop row (max 100 rows)");
165 addCommandDescriptor(RandomScanWithRange1000Test.class, "scanRange1000",
166 "Run random seek scan with both start and stop row (max 1000 rows)");
167 addCommandDescriptor(RandomScanWithRange10000Test.class, "scanRange10000",
168 "Run random seek scan with both start and stop row (max 10000 rows)");
169 addCommandDescriptor(RandomWriteTest.class, "randomWrite",
170 "Run random write test");
171 addCommandDescriptor(SequentialReadTest.class, "sequentialRead",
172 "Run sequential read test");
173 addCommandDescriptor(SequentialWriteTest.class, "sequentialWrite",
174 "Run sequential write test");
175 addCommandDescriptor(ScanTest.class, "scan",
176 "Run scan test (read every row)");
177 addCommandDescriptor(FilteredScanTest.class, "filterScan",
178 "Run scan test using a filter to find a specific row based on it's value " +
179 "(make sure to use --rows=20)");
180 addCommandDescriptor(IncrementTest.class, "increment",
181 "Increment on each row; clients overlap on keyspace so some concurrent operations");
182 addCommandDescriptor(AppendTest.class, "append",
183 "Append on each row; clients overlap on keyspace so some concurrent operations");
184 addCommandDescriptor(CheckAndMutateTest.class, "checkAndMutate",
185 "CheckAndMutate on each row; clients overlap on keyspace so some concurrent operations");
186 addCommandDescriptor(CheckAndPutTest.class, "checkAndPut",
187 "CheckAndPut on each row; clients overlap on keyspace so some concurrent operations");
188 addCommandDescriptor(CheckAndDeleteTest.class, "checkAndDelete",
189 "CheckAndDelete on each row; clients overlap on keyspace so some concurrent operations");
190 }
191
192
193
194
195
196 protected static enum Counter {
197
198 ELAPSED_TIME,
199
200 ROWS
201 }
202
203 protected static class RunResult implements Comparable<RunResult> {
204 public RunResult(long duration, Histogram hist) {
205 this.duration = duration;
206 this.hist = hist;
207 }
208
209 public final long duration;
210 public final Histogram hist;
211
212 @Override
213 public String toString() {
214 return Long.toString(duration);
215 }
216
217 @Override public int compareTo(RunResult o) {
218 return Long.compare(this.duration, o.duration);
219 }
220 }
221
222
223
224
225
226 public PerformanceEvaluation(final Configuration conf) {
227 super(conf);
228 }
229
230 protected static void addCommandDescriptor(Class<? extends Test> cmdClass,
231 String name, String description) {
232 CmdDescriptor cmdDescriptor = new CmdDescriptor(cmdClass, name, description);
233 COMMANDS.put(name, cmdDescriptor);
234 }
235
236
237
238
239 interface Status {
240
241
242
243
244
245 void setStatus(final String msg) throws IOException;
246 }
247
248
249
250
251 public static class EvaluationMapTask
252 extends Mapper<LongWritable, Text, LongWritable, LongWritable> {
253
254
255 public final static String CMD_KEY = "EvaluationMapTask.command";
256
257 public static final String PE_KEY = "EvaluationMapTask.performanceEvalImpl";
258
259 private Class<? extends Test> cmd;
260
261 @Override
262 protected void setup(Context context) throws IOException, InterruptedException {
263 this.cmd = forName(context.getConfiguration().get(CMD_KEY), Test.class);
264
265
266
267 Class<? extends PerformanceEvaluation> peClass =
268 forName(context.getConfiguration().get(PE_KEY), PerformanceEvaluation.class);
269 try {
270 peClass.getConstructor(Configuration.class).newInstance(context.getConfiguration());
271 } catch (Exception e) {
272 throw new IllegalStateException("Could not instantiate PE instance", e);
273 }
274 }
275
276 private <Type> Class<? extends Type> forName(String className, Class<Type> type) {
277 try {
278 return Class.forName(className).asSubclass(type);
279 } catch (ClassNotFoundException e) {
280 throw new IllegalStateException("Could not find class for name: " + className, e);
281 }
282 }
283
284 @Override
285 protected void map(LongWritable key, Text value, final Context context)
286 throws IOException, InterruptedException {
287
288 Status status = new Status() {
289 @Override
290 public void setStatus(String msg) {
291 context.setStatus(msg);
292 }
293 };
294
295 ObjectMapper mapper = new ObjectMapper();
296 TestOptions opts = mapper.readValue(value.toString(), TestOptions.class);
297 Configuration conf = HBaseConfiguration.create(context.getConfiguration());
298 final Connection con = ConnectionFactory.createConnection(conf);
299
300
301 RunResult result = PerformanceEvaluation.runOneClient(this.cmd, conf, con, opts, status);
302
303
304 context.getCounter(Counter.ELAPSED_TIME).increment(result.duration);
305 context.getCounter(Counter.ROWS).increment(opts.perClientRunRows);
306 context.write(new LongWritable(opts.startRow), new LongWritable(result.duration));
307 context.progress();
308 }
309 }
310
311
312
313
314
315
316 static boolean checkTable(Admin admin, TestOptions opts) throws IOException {
317 TableName tableName = TableName.valueOf(opts.tableName);
318 boolean needsDelete = false, exists = admin.tableExists(tableName);
319 boolean isReadCmd = opts.cmdName.toLowerCase(Locale.ROOT).contains("read")
320 || opts.cmdName.toLowerCase(Locale.ROOT).contains("scan");
321 if (!exists && isReadCmd) {
322 throw new IllegalStateException(
323 "Must specify an existing table for read commands. Run a write command first.");
324 }
325 HTableDescriptor desc =
326 exists ? admin.getTableDescriptor(TableName.valueOf(opts.tableName)) : null;
327 byte[][] splits = getSplits(opts);
328
329
330
331 if ((exists && opts.presplitRegions != DEFAULT_OPTS.presplitRegions)
332 || (!isReadCmd && desc != null &&
333 !StringUtils.equals(desc.getRegionSplitPolicyClassName(), opts.splitPolicy))
334 || (!isReadCmd && desc != null && desc.getRegionReplication() != opts.replicas)) {
335 needsDelete = true;
336
337 LOG.debug(Objects.toStringHelper("needsDelete")
338 .add("needsDelete", needsDelete)
339 .add("isReadCmd", isReadCmd)
340 .add("exists", exists)
341 .add("desc", desc)
342 .add("presplit", opts.presplitRegions)
343 .add("splitPolicy", opts.splitPolicy)
344 .add("replicas", opts.replicas));
345 }
346
347
348 if (needsDelete) {
349 if (admin.isTableEnabled(tableName)) {
350 admin.disableTable(tableName);
351 }
352 admin.deleteTable(tableName);
353 }
354
355
356 if (!exists || needsDelete) {
357 desc = getTableDescriptor(opts);
358 if (splits != null) {
359 if (LOG.isDebugEnabled()) {
360 for (int i = 0; i < splits.length; i++) {
361 LOG.debug(" split " + i + ": " + Bytes.toStringBinary(splits[i]));
362 }
363 }
364 }
365 admin.createTable(desc, splits);
366 LOG.info("Table " + desc + " created");
367 }
368 return admin.tableExists(tableName);
369 }
370
371
372
373
374 protected static HTableDescriptor getTableDescriptor(TestOptions opts) {
375 HTableDescriptor tableDesc = new HTableDescriptor(TableName.valueOf(opts.tableName));
376 for (int family = 0; family < opts.families; family++) {
377 byte[] familyName = Bytes.toBytes(FAMILY_NAME_BASE + family);
378 HColumnDescriptor familyDesc = new HColumnDescriptor(familyName);
379 familyDesc.setDataBlockEncoding(opts.blockEncoding);
380 familyDesc.setCompressionType(opts.compression);
381 familyDesc.setBloomFilterType(opts.bloomType);
382 familyDesc.setBlocksize(opts.blockSize);
383 if (opts.inMemoryCF) {
384 familyDesc.setInMemory(true);
385 }
386 tableDesc.addFamily(familyDesc);
387 }
388 if (opts.replicas != DEFAULT_OPTS.replicas) {
389 tableDesc.setRegionReplication(opts.replicas);
390 }
391 if (opts.splitPolicy != null && !opts.splitPolicy.equals(DEFAULT_OPTS.splitPolicy)) {
392 tableDesc.setRegionSplitPolicyClassName(opts.splitPolicy);
393 }
394 return tableDesc;
395 }
396
397
398
399
400 protected static byte[][] getSplits(TestOptions opts) {
401 if (opts.presplitRegions == DEFAULT_OPTS.presplitRegions)
402 return null;
403
404 int numSplitPoints = opts.presplitRegions - 1;
405 byte[][] splits = new byte[numSplitPoints][];
406 int jump = opts.totalRows / opts.presplitRegions;
407 for (int i = 0; i < numSplitPoints; i++) {
408 int rowkey = jump * (1 + i);
409 splits[i] = format(rowkey);
410 }
411 return splits;
412 }
413
414
415
416
417 static RunResult[] doLocalClients(final TestOptions opts, final Configuration conf)
418 throws IOException, InterruptedException {
419 final Class<? extends Test> cmd = determineCommandClass(opts.cmdName);
420 assert cmd != null;
421 @SuppressWarnings("unchecked")
422 Future<RunResult>[] threads = new Future[opts.numClientThreads];
423 RunResult[] results = new RunResult[opts.numClientThreads];
424 ExecutorService pool = Executors.newFixedThreadPool(opts.numClientThreads,
425 new ThreadFactoryBuilder().setNameFormat("TestClient-%s").build());
426 final Connection con = ConnectionFactory.createConnection(conf);
427 for (int i = 0; i < threads.length; i++) {
428 final int index = i;
429 threads[i] = pool.submit(new Callable<RunResult>() {
430 @Override
431 public RunResult call() throws Exception {
432 TestOptions threadOpts = new TestOptions(opts);
433 if (threadOpts.startRow == 0) threadOpts.startRow = index * threadOpts.perClientRunRows;
434 RunResult run = runOneClient(cmd, conf, con, threadOpts, new Status() {
435 @Override
436 public void setStatus(final String msg) throws IOException {
437 LOG.info(msg);
438 }
439 });
440 LOG.info("Finished " + Thread.currentThread().getName() + " in " + run.duration +
441 "ms over " + threadOpts.perClientRunRows + " rows");
442 return run;
443 }
444 });
445 }
446 pool.shutdown();
447
448 for (int i = 0; i < threads.length; i++) {
449 try {
450 results[i] = threads[i].get();
451 } catch (ExecutionException e) {
452 throw new IOException(e.getCause());
453 }
454 }
455 final String test = cmd.getSimpleName();
456 LOG.info("[" + test + "] Summary of timings (ms): "
457 + Arrays.toString(results));
458 Arrays.sort(results);
459 long total = 0;
460 for (RunResult result : results) {
461 total += result.duration;
462 }
463 LOG.info("[" + test + "]"
464 + "\tMin: " + results[0] + "ms"
465 + "\tMax: " + results[results.length - 1] + "ms"
466 + "\tAvg: " + (total / results.length) + "ms");
467
468 con.close();
469
470 return results;
471 }
472
473
474
475
476
477
478
479
480 static Job doMapReduce(TestOptions opts, final Configuration conf)
481 throws IOException, InterruptedException, ClassNotFoundException {
482 final Class<? extends Test> cmd = determineCommandClass(opts.cmdName);
483 assert cmd != null;
484 Path inputDir = writeInputFile(conf, opts);
485 conf.set(EvaluationMapTask.CMD_KEY, cmd.getName());
486 conf.set(EvaluationMapTask.PE_KEY, PerformanceEvaluation.class.getName());
487 Job job = Job.getInstance(conf);
488 job.setJarByClass(PerformanceEvaluation.class);
489 job.setJobName("HBase Performance Evaluation - " + opts.cmdName);
490
491 job.setInputFormatClass(NLineInputFormat.class);
492 NLineInputFormat.setInputPaths(job, inputDir);
493
494 NLineInputFormat.setNumLinesPerSplit(job, 1);
495
496 job.setOutputKeyClass(LongWritable.class);
497 job.setOutputValueClass(LongWritable.class);
498
499 job.setMapperClass(EvaluationMapTask.class);
500 job.setReducerClass(LongSumReducer.class);
501
502 job.setNumReduceTasks(1);
503
504 job.setOutputFormatClass(TextOutputFormat.class);
505 TextOutputFormat.setOutputPath(job, new Path(inputDir.getParent(), "outputs"));
506
507 TableMapReduceUtil.addDependencyJars(job);
508 TableMapReduceUtil.addDependencyJarsForClasses(job.getConfiguration(),
509 Histogram.class,
510 ObjectMapper.class);
511
512 TableMapReduceUtil.initCredentials(job);
513
514 job.waitForCompletion(true);
515 return job;
516 }
517
518
519
520
521
522
523
524 private static Path writeInputFile(final Configuration c, final TestOptions opts) throws IOException {
525 SimpleDateFormat formatter = new SimpleDateFormat("yyyyMMddHHmmss");
526 Path jobdir = new Path(PERF_EVAL_DIR, formatter.format(new Date()));
527 Path inputDir = new Path(jobdir, "inputs");
528
529 FileSystem fs = FileSystem.get(c);
530 fs.mkdirs(inputDir);
531
532 Path inputFile = new Path(inputDir, "input.txt");
533 PrintStream out = new PrintStream(fs.create(inputFile));
534
535 Map<Integer, String> m = new TreeMap<Integer, String>();
536 Hash h = MurmurHash.getInstance();
537 int perClientRows = (opts.totalRows / opts.numClientThreads);
538 try {
539 for (int i = 0; i < 10; i++) {
540 for (int j = 0; j < opts.numClientThreads; j++) {
541 TestOptions next = new TestOptions(opts);
542 next.startRow = (j * perClientRows) + (i * (perClientRows/10));
543 next.perClientRunRows = perClientRows / 10;
544 String s = MAPPER.writeValueAsString(next);
545 LOG.info("maptask input=" + s);
546 int hash = h.hash(Bytes.toBytes(s));
547 m.put(hash, s);
548 }
549 }
550 for (Map.Entry<Integer, String> e: m.entrySet()) {
551 out.println(e.getValue());
552 }
553 } finally {
554 out.close();
555 }
556 return inputDir;
557 }
558
559
560
561
562 static class CmdDescriptor {
563 private Class<? extends Test> cmdClass;
564 private String name;
565 private String description;
566
567 CmdDescriptor(Class<? extends Test> cmdClass, String name, String description) {
568 this.cmdClass = cmdClass;
569 this.name = name;
570 this.description = description;
571 }
572
573 public Class<? extends Test> getCmdClass() {
574 return cmdClass;
575 }
576
577 public String getName() {
578 return name;
579 }
580
581 public String getDescription() {
582 return description;
583 }
584 }
585
586
587
588
589
590
591
592
593 static class TestOptions {
594 String cmdName = null;
595 boolean nomapred = false;
596 boolean filterAll = false;
597 int startRow = 0;
598 float size = 1.0f;
599 int perClientRunRows = DEFAULT_ROWS_PER_GB;
600 int numClientThreads = 1;
601 int totalRows = DEFAULT_ROWS_PER_GB;
602 float sampleRate = 1.0f;
603 double traceRate = 0.0;
604 String tableName = TABLE_NAME;
605 boolean flushCommits = true;
606 boolean writeToWAL = true;
607 boolean autoFlush = false;
608 boolean oneCon = false;
609 boolean useTags = false;
610 int noOfTags = 1;
611 boolean reportLatency = false;
612 int multiGet = 0;
613 int randomSleep = 0;
614 boolean inMemoryCF = false;
615 int presplitRegions = 0;
616 int replicas = HTableDescriptor.DEFAULT_REGION_REPLICATION;
617 String splitPolicy = null;
618 Compression.Algorithm compression = Compression.Algorithm.NONE;
619 BloomType bloomType = BloomType.ROW;
620 int blockSize = HConstants.DEFAULT_BLOCKSIZE;
621 DataBlockEncoding blockEncoding = DataBlockEncoding.NONE;
622 boolean valueRandom = false;
623 boolean valueZipf = false;
624 int valueSize = DEFAULT_VALUE_LENGTH;
625 int period = (this.perClientRunRows / 10) == 0? perClientRunRows: perClientRunRows / 10;
626 int columns = 1;
627 int families = 1;
628 int caching = 30;
629 boolean addColumns = true;
630 long bufferSize = 2l * 1024l * 1024l;
631
632 public TestOptions() {}
633
634
635
636
637
638 public TestOptions(TestOptions that) {
639 this.cmdName = that.cmdName;
640 this.nomapred = that.nomapred;
641 this.startRow = that.startRow;
642 this.size = that.size;
643 this.perClientRunRows = that.perClientRunRows;
644 this.numClientThreads = that.numClientThreads;
645 this.totalRows = that.totalRows;
646 this.sampleRate = that.sampleRate;
647 this.traceRate = that.traceRate;
648 this.tableName = that.tableName;
649 this.flushCommits = that.flushCommits;
650 this.writeToWAL = that.writeToWAL;
651 this.autoFlush = that.autoFlush;
652 this.oneCon = that.oneCon;
653 this.useTags = that.useTags;
654 this.noOfTags = that.noOfTags;
655 this.reportLatency = that.reportLatency;
656 this.multiGet = that.multiGet;
657 this.inMemoryCF = that.inMemoryCF;
658 this.presplitRegions = that.presplitRegions;
659 this.replicas = that.replicas;
660 this.splitPolicy = that.splitPolicy;
661 this.compression = that.compression;
662 this.blockEncoding = that.blockEncoding;
663 this.filterAll = that.filterAll;
664 this.bloomType = that.bloomType;
665 this.blockSize = that.blockSize;
666 this.valueRandom = that.valueRandom;
667 this.valueZipf = that.valueZipf;
668 this.valueSize = that.valueSize;
669 this.period = that.period;
670 this.randomSleep = that.randomSleep;
671 this.addColumns = that.addColumns;
672 this.columns = that.columns;
673 this.families = that.families;
674 this.caching = that.caching;
675 this.bufferSize = that.bufferSize;
676 }
677
678 public int getCaching() {
679 return this.caching;
680 }
681
682 public void setCaching(final int caching) {
683 this.caching = caching;
684 }
685
686 public int getColumns() {
687 return this.columns;
688 }
689
690 public void setColumns(final int columns) {
691 this.columns = columns;
692 }
693
694 public int getFamilies() {
695 return this.families;
696 }
697
698 public void setFamilies(final int families) {
699 this.families = families;
700 }
701
702 public boolean isValueZipf() {
703 return valueZipf;
704 }
705
706 public void setValueZipf(boolean valueZipf) {
707 this.valueZipf = valueZipf;
708 }
709
710 public String getCmdName() {
711 return cmdName;
712 }
713
714 public void setCmdName(String cmdName) {
715 this.cmdName = cmdName;
716 }
717
718 public int getRandomSleep() {
719 return randomSleep;
720 }
721
722 public void setRandomSleep(int randomSleep) {
723 this.randomSleep = randomSleep;
724 }
725
726 public int getReplicas() {
727 return replicas;
728 }
729
730 public void setReplicas(int replicas) {
731 this.replicas = replicas;
732 }
733
734 public String getSplitPolicy() {
735 return splitPolicy;
736 }
737
738 public void setSplitPolicy(String splitPolicy) {
739 this.splitPolicy = splitPolicy;
740 }
741
742 public void setNomapred(boolean nomapred) {
743 this.nomapred = nomapred;
744 }
745
746 public void setFilterAll(boolean filterAll) {
747 this.filterAll = filterAll;
748 }
749
750 public void setStartRow(int startRow) {
751 this.startRow = startRow;
752 }
753
754 public void setSize(float size) {
755 this.size = size;
756 }
757
758 public void setPerClientRunRows(int perClientRunRows) {
759 this.perClientRunRows = perClientRunRows;
760 }
761
762 public void setNumClientThreads(int numClientThreads) {
763 this.numClientThreads = numClientThreads;
764 }
765
766 public void setTotalRows(int totalRows) {
767 this.totalRows = totalRows;
768 }
769
770 public void setSampleRate(float sampleRate) {
771 this.sampleRate = sampleRate;
772 }
773
774 public void setTraceRate(double traceRate) {
775 this.traceRate = traceRate;
776 }
777
778 public void setTableName(String tableName) {
779 this.tableName = tableName;
780 }
781
782 public void setFlushCommits(boolean flushCommits) {
783 this.flushCommits = flushCommits;
784 }
785
786 public void setWriteToWAL(boolean writeToWAL) {
787 this.writeToWAL = writeToWAL;
788 }
789
790 public void setAutoFlush(boolean autoFlush) {
791 this.autoFlush = autoFlush;
792 }
793
794 public void setOneCon(boolean oneCon) {
795 this.oneCon = oneCon;
796 }
797
798 public void setUseTags(boolean useTags) {
799 this.useTags = useTags;
800 }
801
802 public void setNoOfTags(int noOfTags) {
803 this.noOfTags = noOfTags;
804 }
805
806 public void setReportLatency(boolean reportLatency) {
807 this.reportLatency = reportLatency;
808 }
809
810 public void setMultiGet(int multiGet) {
811 this.multiGet = multiGet;
812 }
813
814 public void setInMemoryCF(boolean inMemoryCF) {
815 this.inMemoryCF = inMemoryCF;
816 }
817
818 public void setPresplitRegions(int presplitRegions) {
819 this.presplitRegions = presplitRegions;
820 }
821
822 public void setCompression(Compression.Algorithm compression) {
823 this.compression = compression;
824 }
825
826 public void setBloomType(BloomType bloomType) {
827 this.bloomType = bloomType;
828 }
829
830 public void setBlockSize(int blockSize) {
831 this.blockSize = blockSize;
832 }
833
834 public void setBlockEncoding(DataBlockEncoding blockEncoding) {
835 this.blockEncoding = blockEncoding;
836 }
837
838 public void setValueRandom(boolean valueRandom) {
839 this.valueRandom = valueRandom;
840 }
841
842 public void setValueSize(int valueSize) {
843 this.valueSize = valueSize;
844 }
845
846 public void setBufferSize(long bufferSize) {
847 this.bufferSize = bufferSize;
848 }
849
850 public long getBufferSize() {
851 return this.bufferSize;
852 }
853
854 public void setPeriod(int period) {
855 this.period = period;
856 }
857
858 public boolean isNomapred() {
859 return nomapred;
860 }
861
862 public boolean isFilterAll() {
863 return filterAll;
864 }
865
866 public int getStartRow() {
867 return startRow;
868 }
869
870 public float getSize() {
871 return size;
872 }
873
874 public int getPerClientRunRows() {
875 return perClientRunRows;
876 }
877
878 public int getNumClientThreads() {
879 return numClientThreads;
880 }
881
882 public int getTotalRows() {
883 return totalRows;
884 }
885
886 public float getSampleRate() {
887 return sampleRate;
888 }
889
890 public double getTraceRate() {
891 return traceRate;
892 }
893
894 public String getTableName() {
895 return tableName;
896 }
897
898 public boolean isFlushCommits() {
899 return flushCommits;
900 }
901
902 public boolean isWriteToWAL() {
903 return writeToWAL;
904 }
905
906 public boolean isAutoFlush() {
907 return autoFlush;
908 }
909
910 public boolean isUseTags() {
911 return useTags;
912 }
913
914 public int getNoOfTags() {
915 return noOfTags;
916 }
917
918 public boolean isReportLatency() {
919 return reportLatency;
920 }
921
922 public int getMultiGet() {
923 return multiGet;
924 }
925
926 public boolean isInMemoryCF() {
927 return inMemoryCF;
928 }
929
930 public int getPresplitRegions() {
931 return presplitRegions;
932 }
933
934 public Compression.Algorithm getCompression() {
935 return compression;
936 }
937
938 public DataBlockEncoding getBlockEncoding() {
939 return blockEncoding;
940 }
941
942 public boolean isValueRandom() {
943 return valueRandom;
944 }
945
946 public int getValueSize() {
947 return valueSize;
948 }
949
950 public int getPeriod() {
951 return period;
952 }
953
954 public BloomType getBloomType() {
955 return bloomType;
956 }
957
958 public int getBlockSize() {
959 return blockSize;
960 }
961
962 public boolean isOneCon() {
963 return oneCon;
964 }
965
966 public boolean getAddColumns() {
967 return addColumns;
968 }
969
970 public void setAddColumns(boolean addColumns) {
971 this.addColumns = addColumns;
972 }
973 }
974
975
976
977
978
979 static abstract class Test {
980
981
982 private static final Random randomSeed = new Random(System.currentTimeMillis());
983
984 private static long nextRandomSeed() {
985 return randomSeed.nextLong();
986 }
987 private final int everyN;
988
989 protected final Random rand = new Random(nextRandomSeed());
990 protected final Configuration conf;
991 protected final TestOptions opts;
992
993 private final Status status;
994 private final Sampler<?> traceSampler;
995 private final SpanReceiverHost receiverHost;
996 protected Connection connection;
997
998 private String testName;
999 private Histogram latencyHistogram;
1000 private Histogram valueSizeHistogram;
1001 private Histogram rpcCallsHistogram;
1002 private Histogram remoteRpcCallsHistogram;
1003 private Histogram millisBetweenNextHistogram;
1004 private Histogram regionsScannedHistogram;
1005 private Histogram bytesInResultsHistogram;
1006 private Histogram bytesInRemoteResultsHistogram;
1007 private RandomDistribution.Zipf zipf;
1008
1009
1010
1011
1012
1013 Test(final Connection con, final TestOptions options, final Status status) {
1014 this.connection = con;
1015 this.conf = con == null ? HBaseConfiguration.create() : this.connection.getConfiguration();
1016 this.opts = options;
1017 this.status = status;
1018 this.testName = this.getClass().getSimpleName();
1019 receiverHost = SpanReceiverHost.getInstance(conf);
1020 if (options.traceRate >= 1.0) {
1021 this.traceSampler = Sampler.ALWAYS;
1022 } else if (options.traceRate > 0.0) {
1023 conf.setDouble("hbase.sampler.fraction", options.traceRate);
1024 this.traceSampler = new ProbabilitySampler(new HBaseHTraceConfiguration(conf));
1025 } else {
1026 this.traceSampler = Sampler.NEVER;
1027 }
1028 everyN = (int) (opts.totalRows / (opts.totalRows * opts.sampleRate));
1029 if (options.isValueZipf()) {
1030 this.zipf = new RandomDistribution.Zipf(this.rand, 1, options.getValueSize(), 1.1);
1031 }
1032 LOG.info("Sampling 1 every " + everyN + " out of " + opts.perClientRunRows + " total rows.");
1033 }
1034
1035 int getValueLength(final Random r) {
1036 if (this.opts.isValueRandom()) {
1037 return r.nextInt(opts.valueSize);
1038 } else if (this.opts.isValueZipf()) {
1039 return Math.abs(this.zipf.nextInt());
1040 } else {
1041 return opts.valueSize;
1042 }
1043 }
1044
1045 void updateValueSize(final Result [] rs) throws IOException {
1046 if (rs == null || !isRandomValueSize()) return;
1047 for (Result r: rs) updateValueSize(r);
1048 }
1049
1050 void updateValueSize(final Result r) throws IOException {
1051 if (r == null || !isRandomValueSize()) return;
1052 int size = 0;
1053 for (CellScanner scanner = r.cellScanner(); scanner.advance();) {
1054 size += scanner.current().getValueLength();
1055 }
1056 updateValueSize(size);
1057 }
1058
1059 void updateValueSize(final int valueSize) {
1060 if (!isRandomValueSize()) return;
1061 this.valueSizeHistogram.update(valueSize);
1062 }
1063
1064 void updateScanMetrics(final ScanMetrics metrics) {
1065 if (metrics == null) return;
1066 Map<String,Long> metricsMap = metrics.getMetricsMap();
1067 Long rpcCalls = metricsMap.get(ScanMetrics.RPC_CALLS_METRIC_NAME);
1068 if (rpcCalls != null) {
1069 this.rpcCallsHistogram.update(rpcCalls.longValue());
1070 }
1071 Long remoteRpcCalls = metricsMap.get(ScanMetrics.REMOTE_RPC_CALLS_METRIC_NAME);
1072 if (remoteRpcCalls != null) {
1073 this.remoteRpcCallsHistogram.update(remoteRpcCalls.longValue());
1074 }
1075 Long millisBetweenNext = metricsMap.get(ScanMetrics.MILLIS_BETWEEN_NEXTS_METRIC_NAME);
1076 if (millisBetweenNext != null) {
1077 this.millisBetweenNextHistogram.update(millisBetweenNext.longValue());
1078 }
1079 Long regionsScanned = metricsMap.get(ScanMetrics.REGIONS_SCANNED_METRIC_NAME);
1080 if (regionsScanned != null) {
1081 this.regionsScannedHistogram.update(regionsScanned.longValue());
1082 }
1083 Long bytesInResults = metricsMap.get(ScanMetrics.BYTES_IN_RESULTS_METRIC_NAME);
1084 if (bytesInResults != null && bytesInResults.longValue() > 0) {
1085 this.bytesInResultsHistogram.update(bytesInResults.longValue());
1086 }
1087 Long bytesInRemoteResults = metricsMap.get(ScanMetrics.BYTES_IN_REMOTE_RESULTS_METRIC_NAME);
1088 if (bytesInRemoteResults != null && bytesInRemoteResults.longValue() > 0) {
1089 this.bytesInRemoteResultsHistogram.update(bytesInRemoteResults.longValue());
1090 }
1091 }
1092
1093 String generateStatus(final int sr, final int i, final int lr) {
1094 return sr + "/" + i + "/" + lr + ", latency " + getShortLatencyReport() +
1095 (!isRandomValueSize()? "": ", value size " + getShortValueSizeReport());
1096 }
1097
1098 boolean isRandomValueSize() {
1099 return opts.valueRandom;
1100 }
1101
1102 protected int getReportingPeriod() {
1103 return opts.period;
1104 }
1105
1106
1107
1108
1109 public Histogram getLatencyHistogram() {
1110 return latencyHistogram;
1111 }
1112
1113 void testSetup() throws IOException {
1114
1115 latencyHistogram = YammerHistogramUtils.newHistogram(new UniformSample(1024 * 500));
1116 valueSizeHistogram = YammerHistogramUtils.newHistogram(new UniformSample(1024 * 500));
1117
1118 rpcCallsHistogram = YammerHistogramUtils.newHistogram(new UniformSample(1024 * 500));
1119 remoteRpcCallsHistogram = YammerHistogramUtils.newHistogram(new UniformSample(1024 * 500));
1120 millisBetweenNextHistogram = YammerHistogramUtils.newHistogram(new UniformSample(1024 * 500));
1121 regionsScannedHistogram = YammerHistogramUtils.newHistogram(new UniformSample(1024 * 500));
1122 bytesInResultsHistogram = YammerHistogramUtils.newHistogram(new UniformSample(1024 * 500));
1123 bytesInRemoteResultsHistogram = YammerHistogramUtils.newHistogram(new UniformSample(1024 * 500));
1124
1125 if (!opts.oneCon) {
1126 this.connection = ConnectionFactory.createConnection(conf);
1127 }
1128
1129 onStartup();
1130 }
1131
1132 abstract void onStartup() throws IOException;
1133
1134 void testTakedown() throws IOException {
1135 onTakedown();
1136
1137
1138
1139 synchronized (Test.class) {
1140 status.setStatus("Test : " + testName + ", Thread : " + Thread.currentThread().getName());
1141 status.setStatus("Latency (us) : " + YammerHistogramUtils.getHistogramReport(
1142 latencyHistogram));
1143 status.setStatus("Num measures (latency) : " + latencyHistogram.count());
1144 status.setStatus(YammerHistogramUtils.getPrettyHistogramReport(latencyHistogram));
1145 status.setStatus("ValueSize (bytes) : "
1146 + YammerHistogramUtils.getHistogramReport(valueSizeHistogram));
1147 status.setStatus("Num measures (ValueSize): " + valueSizeHistogram.count());
1148 status.setStatus(YammerHistogramUtils.getPrettyHistogramReport(valueSizeHistogram));
1149 if (rpcCallsHistogram.count() > 0) {
1150 status.setStatus("rpcCalls (count): " +
1151 YammerHistogramUtils.getHistogramReport(rpcCallsHistogram));
1152 }
1153 if (remoteRpcCallsHistogram.count() > 0) {
1154 status.setStatus("remoteRpcCalls (count): " +
1155 YammerHistogramUtils.getHistogramReport(remoteRpcCallsHistogram));
1156 }
1157 if (millisBetweenNextHistogram.count() > 0) {
1158 status.setStatus("millisBetweenNext (latency): " +
1159 YammerHistogramUtils.getHistogramReport(millisBetweenNextHistogram));
1160 }
1161 if (regionsScannedHistogram.count() > 0) {
1162 status.setStatus("regionsScanned (count): " +
1163 YammerHistogramUtils.getHistogramReport(regionsScannedHistogram));
1164 }
1165 if (bytesInResultsHistogram.count() > 0) {
1166 status.setStatus("bytesInResults (size): " +
1167 YammerHistogramUtils.getHistogramReport(bytesInResultsHistogram));
1168 }
1169 if (bytesInRemoteResultsHistogram.count() > 0) {
1170 status.setStatus("bytesInRemoteResults (size): " +
1171 YammerHistogramUtils.getHistogramReport(bytesInRemoteResultsHistogram));
1172 }
1173 }
1174 if (!opts.oneCon) {
1175 connection.close();
1176 }
1177 receiverHost.closeReceivers();
1178 }
1179
1180 abstract void onTakedown() throws IOException;
1181
1182
1183
1184
1185
1186
1187 long test() throws IOException, InterruptedException {
1188 testSetup();
1189 LOG.info("Timed test starting in thread " + Thread.currentThread().getName());
1190 final long startTime = System.nanoTime();
1191 try {
1192 testTimed();
1193 } finally {
1194 testTakedown();
1195 }
1196 return (System.nanoTime() - startTime) / 1000000;
1197 }
1198
1199 int getStartRow() {
1200 return opts.startRow;
1201 }
1202
1203 int getLastRow() {
1204 return getStartRow() + opts.perClientRunRows;
1205 }
1206
1207
1208
1209
1210 void testTimed() throws IOException, InterruptedException {
1211 int startRow = getStartRow();
1212 int lastRow = getLastRow();
1213
1214 for (int i = startRow; i < lastRow; i++) {
1215 if (i % everyN != 0) continue;
1216 long startTime = System.nanoTime();
1217 TraceScope scope = Trace.startSpan("test row", traceSampler);
1218 try {
1219 testRow(i);
1220 } finally {
1221 scope.close();
1222 }
1223
1224
1225
1226 if (opts.multiGet == 0 || (i - startRow + 1) % opts.multiGet == 0) {
1227 latencyHistogram.update((System.nanoTime() - startTime) / 1000);
1228 }
1229 if (status != null && i > 0 && (i % getReportingPeriod()) == 0) {
1230 status.setStatus(generateStatus(startRow, i, lastRow));
1231 }
1232 }
1233 }
1234
1235
1236
1237
1238 public String getShortLatencyReport() {
1239 return YammerHistogramUtils.getShortHistogramReport(this.latencyHistogram);
1240 }
1241
1242
1243
1244
1245 public String getShortValueSizeReport() {
1246 return YammerHistogramUtils.getShortHistogramReport(this.valueSizeHistogram);
1247 }
1248
1249
1250
1251
1252
1253 abstract void testRow(final int i) throws IOException, InterruptedException;
1254 }
1255
1256 static abstract class TableTest extends Test {
1257 protected Table table;
1258
1259 TableTest(Connection con, TestOptions options, Status status) {
1260 super(con, options, status);
1261 }
1262
1263 @Override
1264 void onStartup() throws IOException {
1265 this.table = connection.getTable(TableName.valueOf(opts.tableName));
1266 }
1267
1268 @Override
1269 void onTakedown() throws IOException {
1270 table.close();
1271 }
1272 }
1273
1274 static abstract class BufferedMutatorTest extends Test {
1275 protected BufferedMutator mutator;
1276
1277 BufferedMutatorTest(Connection con, TestOptions options, Status status) {
1278 super(con, options, status);
1279 }
1280
1281 @Override
1282 void onStartup() throws IOException {
1283 BufferedMutatorParams p = new BufferedMutatorParams(TableName.valueOf(opts.tableName));
1284 p.writeBufferSize(opts.bufferSize);
1285 this.mutator = connection.getBufferedMutator(p);
1286 }
1287
1288 @Override
1289 void onTakedown() throws IOException {
1290 mutator.close();
1291 }
1292 }
1293
1294 static class RandomSeekScanTest extends TableTest {
1295 RandomSeekScanTest(Connection con, TestOptions options, Status status) {
1296 super(con, options, status);
1297 }
1298
1299 @Override
1300 void testRow(final int i) throws IOException {
1301 Scan scan = new Scan(getRandomRow(this.rand, opts.totalRows));
1302 FilterList list = new FilterList();
1303 for (int family = 0; family < opts.families; family++) {
1304 byte[] familyName = Bytes.toBytes(FAMILY_NAME_BASE + family);
1305 if (opts.addColumns) {
1306 for (int column = 0; column < opts.columns; column++) {
1307 byte [] qualifier = column == 0? COLUMN_ZERO: Bytes.toBytes("" + column);
1308 scan.addColumn(familyName, qualifier);
1309 }
1310 } else {
1311 scan.addFamily(familyName);
1312 }
1313 }
1314 if (opts.filterAll) {
1315 list.addFilter(new FilterAllFilter());
1316 }
1317 list.addFilter(new WhileMatchFilter(new PageFilter(120)));
1318 scan.setFilter(list);
1319 scan.setCaching(opts.caching);
1320 scan.setScanMetricsEnabled(true);
1321 ResultScanner s = this.table.getScanner(scan);
1322 try {
1323 for (Result rr; (rr = s.next()) != null;) {
1324 updateValueSize(rr);
1325 }
1326 } finally {
1327 if (s != null) {
1328 updateScanMetrics(scan.getScanMetrics());
1329 s.close();
1330 }
1331 }
1332 }
1333
1334 @Override
1335 protected int getReportingPeriod() {
1336 int period = opts.perClientRunRows / 100;
1337 return period == 0 ? opts.perClientRunRows : period;
1338 }
1339
1340 }
1341
1342 static abstract class RandomScanWithRangeTest extends TableTest {
1343 RandomScanWithRangeTest(Connection con, TestOptions options, Status status) {
1344 super(con, options, status);
1345 }
1346
1347 @Override
1348 void testRow(final int i) throws IOException {
1349 Pair<byte[], byte[]> startAndStopRow = getStartAndStopRow();
1350 Scan scan = new Scan(startAndStopRow.getFirst(), startAndStopRow.getSecond());
1351 for (int family = 0; family < opts.families; family++) {
1352 byte[] familyName = Bytes.toBytes(FAMILY_NAME_BASE + family);
1353 if (opts.addColumns) {
1354 for (int column = 0; column < opts.columns; column++) {
1355 byte [] qualifier = column == 0? COLUMN_ZERO: Bytes.toBytes("" + column);
1356 scan.addColumn(familyName, qualifier);
1357 }
1358 } else {
1359 scan.addFamily(familyName);
1360 }
1361 }
1362 if (opts.filterAll) {
1363 scan.setFilter(new FilterAllFilter());
1364 }
1365 scan.setCaching(opts.caching);
1366 scan.setScanMetricsEnabled(true);
1367 Result r = null;
1368 int count = 0;
1369 ResultScanner s = this.table.getScanner(scan);
1370 try {
1371 for (; (r = s.next()) != null;) {
1372 updateValueSize(r);
1373 count++;
1374 }
1375 if (i % 100 == 0) {
1376 LOG.info(String.format("Scan for key range %s - %s returned %s rows",
1377 Bytes.toString(startAndStopRow.getFirst()),
1378 Bytes.toString(startAndStopRow.getSecond()), count));
1379 }
1380 } finally {
1381 if (s != null) {
1382 updateScanMetrics(scan.getScanMetrics());
1383 s.close();
1384 }
1385 }
1386 }
1387
1388 protected abstract Pair<byte[],byte[]> getStartAndStopRow();
1389
1390 protected Pair<byte[], byte[]> generateStartAndStopRows(int maxRange) {
1391 int start = this.rand.nextInt(Integer.MAX_VALUE) % opts.totalRows;
1392 int stop = start + maxRange;
1393 return new Pair<byte[],byte[]>(format(start), format(stop));
1394 }
1395
1396 @Override
1397 protected int getReportingPeriod() {
1398 int period = opts.perClientRunRows / 100;
1399 return period == 0? opts.perClientRunRows: period;
1400 }
1401 }
1402
1403 static class RandomScanWithRange10Test extends RandomScanWithRangeTest {
1404 RandomScanWithRange10Test(Connection con, TestOptions options, Status status) {
1405 super(con, options, status);
1406 }
1407
1408 @Override
1409 protected Pair<byte[], byte[]> getStartAndStopRow() {
1410 return generateStartAndStopRows(10);
1411 }
1412 }
1413
1414 static class RandomScanWithRange100Test extends RandomScanWithRangeTest {
1415 RandomScanWithRange100Test(Connection con, TestOptions options, Status status) {
1416 super(con, options, status);
1417 }
1418
1419 @Override
1420 protected Pair<byte[], byte[]> getStartAndStopRow() {
1421 return generateStartAndStopRows(100);
1422 }
1423 }
1424
1425 static class RandomScanWithRange1000Test extends RandomScanWithRangeTest {
1426 RandomScanWithRange1000Test(Connection con, TestOptions options, Status status) {
1427 super(con, options, status);
1428 }
1429
1430 @Override
1431 protected Pair<byte[], byte[]> getStartAndStopRow() {
1432 return generateStartAndStopRows(1000);
1433 }
1434 }
1435
1436 static class RandomScanWithRange10000Test extends RandomScanWithRangeTest {
1437 RandomScanWithRange10000Test(Connection con, TestOptions options, Status status) {
1438 super(con, options, status);
1439 }
1440
1441 @Override
1442 protected Pair<byte[], byte[]> getStartAndStopRow() {
1443 return generateStartAndStopRows(10000);
1444 }
1445 }
1446
1447 static class RandomReadTest extends TableTest {
1448 private final Consistency consistency;
1449 private ArrayList<Get> gets;
1450 private Random rd = new Random();
1451
1452 RandomReadTest(Connection con, TestOptions options, Status status) {
1453 super(con, options, status);
1454 consistency = options.replicas == DEFAULT_OPTS.replicas ? null : Consistency.TIMELINE;
1455 if (opts.multiGet > 0) {
1456 LOG.info("MultiGet enabled. Sending GETs in batches of " + opts.multiGet + ".");
1457 this.gets = new ArrayList<Get>(opts.multiGet);
1458 }
1459 }
1460
1461 @Override
1462 void testRow(final int i) throws IOException, InterruptedException {
1463 if (opts.randomSleep > 0) {
1464 Thread.sleep(rd.nextInt(opts.randomSleep));
1465 }
1466 Get get = new Get(getRandomRow(this.rand, opts.totalRows));
1467 for (int family = 0; family < opts.families; family++) {
1468 byte[] familyName = Bytes.toBytes(FAMILY_NAME_BASE + family);
1469 if (opts.addColumns) {
1470 for (int column = 0; column < opts.columns; column++) {
1471 byte [] qualifier = column == 0? COLUMN_ZERO: Bytes.toBytes("" + column);
1472 get.addColumn(familyName, qualifier);
1473 }
1474 } else {
1475 get.addFamily(familyName);
1476 }
1477 }
1478 if (opts.filterAll) {
1479 get.setFilter(new FilterAllFilter());
1480 }
1481 get.setConsistency(consistency);
1482 if (LOG.isTraceEnabled()) LOG.trace(get.toString());
1483 if (opts.multiGet > 0) {
1484 this.gets.add(get);
1485 if (this.gets.size() == opts.multiGet) {
1486 Result [] rs = this.table.get(this.gets);
1487 updateValueSize(rs);
1488 this.gets.clear();
1489 }
1490 } else {
1491 updateValueSize(this.table.get(get));
1492 }
1493 }
1494
1495 @Override
1496 protected int getReportingPeriod() {
1497 int period = opts.perClientRunRows / 10;
1498 return period == 0 ? opts.perClientRunRows : period;
1499 }
1500
1501 @Override
1502 protected void testTakedown() throws IOException {
1503 if (this.gets != null && this.gets.size() > 0) {
1504 this.table.get(gets);
1505 this.gets.clear();
1506 }
1507 super.testTakedown();
1508 }
1509 }
1510
1511 static class RandomWriteTest extends BufferedMutatorTest {
1512 RandomWriteTest(Connection con, TestOptions options, Status status) {
1513 super(con, options, status);
1514 }
1515
1516 @Override
1517 void testRow(final int i) throws IOException {
1518 byte[] row = getRandomRow(this.rand, opts.totalRows);
1519 Put put = new Put(row);
1520 for (int family = 0; family < opts.families; family++) {
1521 byte[] familyName = Bytes.toBytes(FAMILY_NAME_BASE + family);
1522 for (int column = 0; column < opts.columns; column++) {
1523 byte [] qualifier = column == 0? COLUMN_ZERO: Bytes.toBytes("" + column);
1524 byte[] value = generateData(this.rand, getValueLength(this.rand));
1525 if (opts.useTags) {
1526 byte[] tag = generateData(this.rand, TAG_LENGTH);
1527 Tag[] tags = new Tag[opts.noOfTags];
1528 for (int n = 0; n < opts.noOfTags; n++) {
1529 Tag t = new Tag((byte) n, tag);
1530 tags[n] = t;
1531 }
1532 KeyValue kv = new KeyValue(row, familyName, qualifier, HConstants.LATEST_TIMESTAMP,
1533 value, tags);
1534 put.add(kv);
1535 updateValueSize(kv.getValueLength());
1536 } else {
1537 put.add(familyName, qualifier, value);
1538 updateValueSize(value.length);
1539 }
1540 }
1541 }
1542 put.setDurability(opts.writeToWAL ? Durability.SYNC_WAL : Durability.SKIP_WAL);
1543 mutator.mutate(put);
1544 if (opts.autoFlush) {
1545 mutator.flush();
1546 }
1547 }
1548 }
1549
1550 static class ScanTest extends TableTest {
1551 private Scan scan;
1552 private ResultScanner testScanner;
1553
1554 ScanTest(Connection con, TestOptions options, Status status) {
1555 super(con, options, status);
1556 }
1557
1558 @Override
1559 void testTakedown() throws IOException {
1560 if (this.testScanner != null) {
1561 this.testScanner.close();
1562 }
1563 super.testTakedown();
1564 }
1565
1566
1567 @Override
1568 void testRow(final int i) throws IOException {
1569 if (this.testScanner == null) {
1570 scan = new Scan(format(opts.startRow));
1571 for (int family = 0; family < opts.families; family++) {
1572 byte[] familyName = Bytes.toBytes(FAMILY_NAME_BASE + family);
1573 if (opts.addColumns) {
1574 for (int column = 0; column < opts.columns; column++) {
1575 byte [] qualifier = column == 0? COLUMN_ZERO: Bytes.toBytes("" + column);
1576 scan.addColumn(familyName, qualifier);
1577 }
1578 } else {
1579 scan.addFamily(familyName);
1580 }
1581 }
1582 if (opts.filterAll) {
1583 scan.setFilter(new FilterAllFilter());
1584 }
1585 scan.setCaching(opts.caching);
1586 scan.setScanMetricsEnabled(true);
1587 this.testScanner = table.getScanner(scan);
1588 }
1589 try {
1590 Result r = testScanner.next();
1591 updateValueSize(r);
1592 } finally {
1593 updateScanMetrics(scan.getScanMetrics());
1594 }
1595 }
1596 }
1597
1598
1599
1600
1601
1602
1603
1604
1605
1606 static abstract class CASTableTest extends TableTest {
1607 private final byte [] qualifier;
1608 CASTableTest(Connection con, TestOptions options, Status status) {
1609 super(con, options, status);
1610 qualifier = Bytes.toBytes(this.getClass().getSimpleName());
1611 }
1612
1613 byte [] getQualifier() {
1614 return this.qualifier;
1615 }
1616
1617 @Override
1618 int getStartRow() {
1619 return 0;
1620 }
1621
1622 @Override
1623 int getLastRow() {
1624 return opts.perClientRunRows;
1625 }
1626 }
1627
1628 static class IncrementTest extends CASTableTest {
1629 IncrementTest(Connection con, TestOptions options, Status status) {
1630 super(con, options, status);
1631 }
1632
1633 @Override
1634 void testRow(final int i) throws IOException {
1635 Increment increment = new Increment(format(i));
1636
1637
1638
1639 for (int family = 0; family < opts.families; family++) {
1640 byte[] familyName = Bytes.toBytes(FAMILY_NAME_BASE + family);
1641 increment.addColumn(familyName, getQualifier(), 1l);
1642 }
1643 updateValueSize(this.table.increment(increment));
1644 }
1645 }
1646
1647 static class AppendTest extends CASTableTest {
1648 AppendTest(Connection con, TestOptions options, Status status) {
1649 super(con, options, status);
1650 }
1651
1652 @Override
1653 void testRow(final int i) throws IOException {
1654 byte [] bytes = format(i);
1655 Append append = new Append(bytes);
1656
1657
1658
1659 for (int family = 0; family < opts.families; family++) {
1660 byte[] familyName = Bytes.toBytes(FAMILY_NAME_BASE + family);
1661 append.add(familyName, getQualifier(), bytes);
1662 }
1663 updateValueSize(this.table.append(append));
1664 }
1665 }
1666
1667 static class CheckAndMutateTest extends CASTableTest {
1668 CheckAndMutateTest(Connection con, TestOptions options, Status status) {
1669 super(con, options, status);
1670 }
1671
1672 @Override
1673 void testRow(final int i) throws IOException {
1674 final byte [] bytes = format(i);
1675
1676
1677 Put put = new Put(bytes);
1678 put.addColumn(FAMILY_ZERO, getQualifier(), bytes);
1679 this.table.put(put);
1680 RowMutations mutations = new RowMutations(bytes);
1681 mutations.add(put);
1682 this.table.checkAndMutate(bytes, FAMILY_ZERO, getQualifier(), CompareOp.EQUAL, bytes,
1683 mutations);
1684 }
1685 }
1686
1687 static class CheckAndPutTest extends CASTableTest {
1688 CheckAndPutTest(Connection con, TestOptions options, Status status) {
1689 super(con, options, status);
1690 }
1691
1692 @Override
1693 void testRow(final int i) throws IOException {
1694 final byte [] bytes = format(i);
1695
1696
1697 Put put = new Put(bytes);
1698 put.addColumn(FAMILY_ZERO, getQualifier(), bytes);
1699 this.table.put(put);
1700 this.table.checkAndPut(bytes, FAMILY_ZERO, getQualifier(), CompareOp.EQUAL, bytes, put);
1701 }
1702 }
1703
1704 static class CheckAndDeleteTest extends CASTableTest {
1705 CheckAndDeleteTest(Connection con, TestOptions options, Status status) {
1706 super(con, options, status);
1707 }
1708
1709 @Override
1710 void testRow(final int i) throws IOException {
1711 final byte [] bytes = format(i);
1712
1713
1714 Put put = new Put(bytes);
1715 put.addColumn(FAMILY_ZERO, getQualifier(), bytes);
1716 this.table.put(put);
1717 Delete delete = new Delete(put.getRow());
1718 delete.addColumn(FAMILY_ZERO, getQualifier());
1719 this.table.checkAndDelete(bytes, FAMILY_ZERO, getQualifier(), CompareOp.EQUAL, bytes, delete);
1720 }
1721 }
1722
1723 static class SequentialReadTest extends TableTest {
1724 SequentialReadTest(Connection con, TestOptions options, Status status) {
1725 super(con, options, status);
1726 }
1727
1728 @Override
1729 void testRow(final int i) throws IOException {
1730 Get get = new Get(format(i));
1731 for (int family = 0; family < opts.families; family++) {
1732 byte[] familyName = Bytes.toBytes(FAMILY_NAME_BASE + family);
1733 if (opts.addColumns) {
1734 for (int column = 0; column < opts.columns; column++) {
1735 byte [] qualifier = column == 0? COLUMN_ZERO: Bytes.toBytes("" + column);
1736 get.addColumn(familyName, qualifier);
1737 }
1738 } else {
1739 get.addFamily(familyName);
1740 }
1741 }
1742 if (opts.filterAll) {
1743 get.setFilter(new FilterAllFilter());
1744 }
1745 updateValueSize(table.get(get));
1746 }
1747 }
1748
1749 static class SequentialWriteTest extends BufferedMutatorTest {
1750 SequentialWriteTest(Connection con, TestOptions options, Status status) {
1751 super(con, options, status);
1752 }
1753
1754 @Override
1755 void testRow(final int i) throws IOException {
1756 byte[] row = format(i);
1757 Put put = new Put(row);
1758 for (int family = 0; family < opts.families; family++) {
1759 byte[] familyName = Bytes.toBytes(FAMILY_NAME_BASE + family);
1760 for (int column = 0; column < opts.columns; column++) {
1761 byte [] qualifier = column == 0? COLUMN_ZERO: Bytes.toBytes("" + column);
1762 byte[] value = generateData(this.rand, getValueLength(this.rand));
1763 if (opts.useTags) {
1764 byte[] tag = generateData(this.rand, TAG_LENGTH);
1765 Tag[] tags = new Tag[opts.noOfTags];
1766 for (int n = 0; n < opts.noOfTags; n++) {
1767 Tag t = new Tag((byte) n, tag);
1768 tags[n] = t;
1769 }
1770 KeyValue kv = new KeyValue(row, familyName, qualifier, HConstants.LATEST_TIMESTAMP,
1771 value, tags);
1772 put.add(kv);
1773 updateValueSize(kv.getValueLength());
1774 } else {
1775 put.add(familyName, qualifier, value);
1776 updateValueSize(value.length);
1777 }
1778 }
1779 }
1780 put.setDurability(opts.writeToWAL ? Durability.SYNC_WAL : Durability.SKIP_WAL);
1781 mutator.mutate(put);
1782 if (opts.autoFlush) {
1783 mutator.flush();
1784 }
1785 }
1786 }
1787
1788 static class FilteredScanTest extends TableTest {
1789 protected static final Log LOG = LogFactory.getLog(FilteredScanTest.class.getName());
1790
1791 FilteredScanTest(Connection con, TestOptions options, Status status) {
1792 super(con, options, status);
1793 if (opts.perClientRunRows == DEFAULT_ROWS_PER_GB) {
1794 LOG.warn("Option \"rows\" unspecified. Using default value " + DEFAULT_ROWS_PER_GB
1795 + ". This could take a very long time.");
1796 }
1797 }
1798
1799 @Override
1800 void testRow(int i) throws IOException {
1801 byte[] value = generateData(this.rand, getValueLength(this.rand));
1802 Scan scan = constructScan(value);
1803 ResultScanner scanner = null;
1804 try {
1805 scanner = this.table.getScanner(scan);
1806 for (Result r = null; (r = scanner.next()) != null;) {
1807 updateValueSize(r);
1808 }
1809 } finally {
1810 if (scanner != null) {
1811 updateScanMetrics(scan.getScanMetrics());
1812 scanner.close();
1813 }
1814 }
1815 }
1816
1817 protected Scan constructScan(byte[] valuePrefix) throws IOException {
1818 FilterList list = new FilterList();
1819 Filter filter = new SingleColumnValueFilter(FAMILY_ZERO, COLUMN_ZERO,
1820 CompareFilter.CompareOp.EQUAL, new BinaryComparator(valuePrefix));
1821 list.addFilter(filter);
1822 if (opts.filterAll) {
1823 list.addFilter(new FilterAllFilter());
1824 }
1825 Scan scan = new Scan();
1826 if (opts.addColumns) {
1827 for (int column = 0; column < opts.columns; column++) {
1828 byte [] qualifier = column == 0? COLUMN_ZERO: Bytes.toBytes("" + column);
1829 scan.addColumn(FAMILY_ZERO, qualifier);
1830 }
1831 } else {
1832 scan.addFamily(FAMILY_ZERO);
1833 }
1834 scan.setFilter(list);
1835 scan.setCaching(opts.caching);
1836 scan.setScanMetricsEnabled(true);
1837 return scan;
1838 }
1839 }
1840
1841
1842
1843
1844
1845
1846
1847 private static String calculateMbps(int rows, long timeMs, final int valueSize, int families,
1848 int columns) {
1849 BigDecimal rowSize = BigDecimal.valueOf(ROW_LENGTH +
1850 ((valueSize + (FAMILY_NAME_BASE.length()+1) + COLUMN_ZERO.length) * columns) * families);
1851 BigDecimal mbps = BigDecimal.valueOf(rows).multiply(rowSize, CXT)
1852 .divide(BigDecimal.valueOf(timeMs), CXT).multiply(MS_PER_SEC, CXT)
1853 .divide(BYTES_PER_MB, CXT);
1854 return FMT.format(mbps) + " MB/s";
1855 }
1856
1857
1858
1859
1860
1861
1862
1863 public static byte [] format(final int number) {
1864 byte [] b = new byte[ROW_LENGTH];
1865 int d = Math.abs(number);
1866 for (int i = b.length - 1; i >= 0; i--) {
1867 b[i] = (byte)((d % 10) + '0');
1868 d /= 10;
1869 }
1870 return b;
1871 }
1872
1873
1874
1875
1876
1877
1878
1879 public static byte[] generateData(final Random r, int length) {
1880 byte [] b = new byte [length];
1881 int i;
1882
1883 for(i = 0; i < (length-8); i += 8) {
1884 b[i] = (byte) (65 + r.nextInt(26));
1885 b[i+1] = b[i];
1886 b[i+2] = b[i];
1887 b[i+3] = b[i];
1888 b[i+4] = b[i];
1889 b[i+5] = b[i];
1890 b[i+6] = b[i];
1891 b[i+7] = b[i];
1892 }
1893
1894 byte a = (byte) (65 + r.nextInt(26));
1895 for(; i < length; i++) {
1896 b[i] = a;
1897 }
1898 return b;
1899 }
1900
1901
1902
1903
1904
1905 @Deprecated
1906 public static byte[] generateValue(final Random r) {
1907 return generateData(r, DEFAULT_VALUE_LENGTH);
1908 }
1909
1910 static byte [] getRandomRow(final Random random, final int totalRows) {
1911 return format(random.nextInt(Integer.MAX_VALUE) % totalRows);
1912 }
1913
1914 static RunResult runOneClient(final Class<? extends Test> cmd, Configuration conf, Connection con,
1915 TestOptions opts, final Status status)
1916 throws IOException, InterruptedException {
1917 status.setStatus("Start " + cmd + " at offset " + opts.startRow + " for " +
1918 opts.perClientRunRows + " rows");
1919 long totalElapsedTime;
1920
1921 final Test t;
1922 try {
1923 Constructor<? extends Test> constructor =
1924 cmd.getDeclaredConstructor(Connection.class, TestOptions.class, Status.class);
1925 t = constructor.newInstance(con, opts, status);
1926 } catch (NoSuchMethodException e) {
1927 throw new IllegalArgumentException("Invalid command class: " +
1928 cmd.getName() + ". It does not provide a constructor as described by " +
1929 "the javadoc comment. Available constructors are: " +
1930 Arrays.toString(cmd.getConstructors()));
1931 } catch (Exception e) {
1932 throw new IllegalStateException("Failed to construct command class", e);
1933 }
1934 totalElapsedTime = t.test();
1935
1936 status.setStatus("Finished " + cmd + " in " + totalElapsedTime +
1937 "ms at offset " + opts.startRow + " for " + opts.perClientRunRows + " rows" +
1938 " (" + calculateMbps((int)(opts.perClientRunRows * opts.sampleRate), totalElapsedTime,
1939 getAverageValueLength(opts), opts.families, opts.columns) + ")");
1940
1941 return new RunResult(totalElapsedTime, t.getLatencyHistogram());
1942 }
1943
1944 private static int getAverageValueLength(final TestOptions opts) {
1945 return opts.valueRandom? opts.valueSize/2: opts.valueSize;
1946 }
1947
1948 private void runTest(final Class<? extends Test> cmd, TestOptions opts) throws IOException,
1949 InterruptedException, ClassNotFoundException {
1950
1951
1952 LOG.info(cmd.getSimpleName() + " test run options=" + MAPPER.writeValueAsString(opts));
1953 try(Connection conn = ConnectionFactory.createConnection(getConf());
1954 Admin admin = conn.getAdmin()) {
1955 checkTable(admin, opts);
1956 }
1957 if (opts.nomapred) {
1958 doLocalClients(opts, getConf());
1959 } else {
1960 doMapReduce(opts, getConf());
1961 }
1962 }
1963
1964 protected void printUsage() {
1965 printUsage(this.getClass().getName(), null);
1966 }
1967
1968 protected static void printUsage(final String message) {
1969 printUsage(PerformanceEvaluation.class.getName(), message);
1970 }
1971
1972 protected static void printUsageAndExit(final String message, final int exitCode) {
1973 printUsage(message);
1974 System.exit(exitCode);
1975 }
1976
1977 protected static void printUsage(final String className, final String message) {
1978 if (message != null && message.length() > 0) {
1979 System.err.println(message);
1980 }
1981 System.err.println("Usage: java " + className + " \\");
1982 System.err.println(" <OPTIONS> [-D<property=value>]* <command> <nclients>");
1983 System.err.println();
1984 System.err.println("Options:");
1985 System.err.println(" nomapred Run multiple clients using threads " +
1986 "(rather than use mapreduce)");
1987 System.err.println(" rows Rows each client runs. Default: " +
1988 DEFAULT_OPTS.getPerClientRunRows());
1989 System.err.println(" size Total size in GiB. Mutually exclusive with --rows. " +
1990 "Default: 1.0.");
1991 System.err.println(" sampleRate Execute test on a sample of total " +
1992 "rows. Only supported by randomRead. Default: 1.0");
1993 System.err.println(" traceRate Enable HTrace spans. Initiate tracing every N rows. " +
1994 "Default: 0");
1995 System.err.println(" table Alternate table name. Default: 'TestTable'");
1996 System.err.println(" multiGet If >0, when doing RandomRead, perform multiple gets " +
1997 "instead of single gets. Default: 0");
1998 System.err.println(" compress Compression type to use (GZ, LZO, ...). Default: 'NONE'");
1999 System.err.println(" flushCommits Used to determine if the test should flush the table. " +
2000 "Default: false");
2001 System.err.println(" writeToWAL Set writeToWAL on puts. Default: True");
2002 System.err.println(" autoFlush Set autoFlush on htable. Default: False");
2003 System.err.println(" oneCon all the threads share the same connection. Default: False");
2004 System.err.println(" presplit Create presplit table. If a table with same name exists,"
2005 + " it'll be deleted and recreated (instead of verifying count of its existing regions). "
2006 + "Recommended for accurate perf analysis (see guide). Default: disabled");
2007 System.err.println(" inmemory Tries to keep the HFiles of the CF " +
2008 "inmemory as far as possible. Not guaranteed that reads are always served " +
2009 "from memory. Default: false");
2010 System.err.println(" usetags Writes tags along with KVs. Use with HFile V3. " +
2011 "Default: false");
2012 System.err.println(" numoftags Specify the no of tags that would be needed. " +
2013 "This works only if usetags is true. Default: " + DEFAULT_OPTS.noOfTags);
2014 System.err.println(" filterAll Helps to filter out all the rows on the server side" +
2015 " there by not returning any thing back to the client. Helps to check the server side" +
2016 " performance. Uses FilterAllFilter internally. ");
2017 System.err.println(" latency Set to report operation latencies. Default: False");
2018 System.err.println(" bloomFilter Bloom filter type, one of " + Arrays.toString(BloomType.values()));
2019 System.err.println(" blockEncoding Block encoding to use. Value should be one of "
2020 + Arrays.toString(DataBlockEncoding.values()) + ". Default: NONE");
2021 System.err.println(" valueSize Pass value size to use: Default: " +
2022 DEFAULT_OPTS.getValueSize());
2023 System.err.println(" valueRandom Set if we should vary value size between 0 and " +
2024 "'valueSize'; set on read for stats on size: Default: Not set.");
2025 System.err.println(" valueZipf Set if we should vary value size between 0 and " +
2026 "'valueSize' in zipf form: Default: Not set.");
2027 System.err.println(" period Report every 'period' rows: " +
2028 "Default: opts.perClientRunRows / 10 = " + DEFAULT_OPTS.getPerClientRunRows()/10);
2029 System.err.println(" multiGet Batch gets together into groups of N. Only supported " +
2030 "by randomRead. Default: disabled");
2031 System.err.println(" blockSize Blocksize to use when writing out hfiles. ");
2032 System.err.println(" addColumns Adds columns to scans/gets explicitly. Default: true");
2033 System.err.println(" replicas Enable region replica testing. Defaults: 1.");
2034 System.err.println(" splitPolicy Specify a custom RegionSplitPolicy for the table.");
2035 System.err.println(" randomSleep Do a random sleep before each get between 0 and entered value. Defaults: 0");
2036 System.err.println(" columns Columns to write per row. Default: 1");
2037 System.err.println(" families Specify number of column families for the table. Default: 1");
2038 System.err.println(" caching Scan caching to use. Default: 30");
2039 System.err.println(" bufferSize Set the value of client side buffering. Default: 2MB");
2040 System.err.println();
2041 System.err.println(" Note: -D properties will be applied to the conf used. ");
2042 System.err.println(" For example: ");
2043 System.err.println(" -Dmapreduce.output.fileoutputformat.compress=true");
2044 System.err.println(" -Dmapreduce.task.timeout=60000");
2045 System.err.println();
2046 System.err.println("Command:");
2047 for (CmdDescriptor command : COMMANDS.values()) {
2048 System.err.println(String.format(" %-15s %s", command.getName(), command.getDescription()));
2049 }
2050 System.err.println();
2051 System.err.println("Args:");
2052 System.err.println(" nclients Integer. Required. Total number of clients "
2053 + "(and HRegionServers) running. 1 <= value <= 500");
2054 System.err.println("Examples:");
2055 System.err.println(" To run a single client doing the default 1M sequentialWrites:");
2056 System.err.println(" $ bin/hbase " + className + " sequentialWrite 1");
2057 System.err.println(" To run 10 clients doing increments over ten rows:");
2058 System.err.println(" $ bin/hbase " + className + " --rows=10 --nomapred increment 10");
2059 }
2060
2061
2062
2063
2064
2065
2066
2067 static TestOptions parseOpts(Queue<String> args) {
2068 TestOptions opts = new TestOptions();
2069
2070 String cmd = null;
2071 while ((cmd = args.poll()) != null) {
2072 if (cmd.equals("-h") || cmd.startsWith("--h")) {
2073
2074 args.add(cmd);
2075 break;
2076 }
2077
2078 final String nmr = "--nomapred";
2079 if (cmd.startsWith(nmr)) {
2080 opts.nomapred = true;
2081 continue;
2082 }
2083
2084 final String rows = "--rows=";
2085 if (cmd.startsWith(rows)) {
2086 opts.perClientRunRows = Integer.parseInt(cmd.substring(rows.length()));
2087 continue;
2088 }
2089
2090 final String sampleRate = "--sampleRate=";
2091 if (cmd.startsWith(sampleRate)) {
2092 opts.sampleRate = Float.parseFloat(cmd.substring(sampleRate.length()));
2093 continue;
2094 }
2095
2096 final String table = "--table=";
2097 if (cmd.startsWith(table)) {
2098 opts.tableName = cmd.substring(table.length());
2099 continue;
2100 }
2101
2102 final String startRow = "--startRow=";
2103 if (cmd.startsWith(startRow)) {
2104 opts.startRow = Integer.parseInt(cmd.substring(startRow.length()));
2105 continue;
2106 }
2107
2108 final String compress = "--compress=";
2109 if (cmd.startsWith(compress)) {
2110 opts.compression = Compression.Algorithm.valueOf(cmd.substring(compress.length()));
2111 continue;
2112 }
2113
2114 final String traceRate = "--traceRate=";
2115 if (cmd.startsWith(traceRate)) {
2116 opts.traceRate = Double.parseDouble(cmd.substring(traceRate.length()));
2117 continue;
2118 }
2119
2120 final String blockEncoding = "--blockEncoding=";
2121 if (cmd.startsWith(blockEncoding)) {
2122 opts.blockEncoding = DataBlockEncoding.valueOf(cmd.substring(blockEncoding.length()));
2123 continue;
2124 }
2125
2126 final String flushCommits = "--flushCommits=";
2127 if (cmd.startsWith(flushCommits)) {
2128 opts.flushCommits = Boolean.parseBoolean(cmd.substring(flushCommits.length()));
2129 continue;
2130 }
2131
2132 final String writeToWAL = "--writeToWAL=";
2133 if (cmd.startsWith(writeToWAL)) {
2134 opts.writeToWAL = Boolean.parseBoolean(cmd.substring(writeToWAL.length()));
2135 continue;
2136 }
2137
2138 final String presplit = "--presplit=";
2139 if (cmd.startsWith(presplit)) {
2140 opts.presplitRegions = Integer.parseInt(cmd.substring(presplit.length()));
2141 continue;
2142 }
2143
2144 final String inMemory = "--inmemory=";
2145 if (cmd.startsWith(inMemory)) {
2146 opts.inMemoryCF = Boolean.parseBoolean(cmd.substring(inMemory.length()));
2147 continue;
2148 }
2149
2150 final String autoFlush = "--autoFlush=";
2151 if (cmd.startsWith(autoFlush)) {
2152 opts.autoFlush = Boolean.parseBoolean(cmd.substring(autoFlush.length()));
2153 continue;
2154 }
2155
2156 final String onceCon = "--oneCon=";
2157 if (cmd.startsWith(onceCon)) {
2158 opts.oneCon = Boolean.parseBoolean(cmd.substring(onceCon.length()));
2159 continue;
2160 }
2161
2162 final String latency = "--latency";
2163 if (cmd.startsWith(latency)) {
2164 opts.reportLatency = true;
2165 continue;
2166 }
2167
2168 final String multiGet = "--multiGet=";
2169 if (cmd.startsWith(multiGet)) {
2170 opts.multiGet = Integer.parseInt(cmd.substring(multiGet.length()));
2171 continue;
2172 }
2173
2174 final String useTags = "--usetags=";
2175 if (cmd.startsWith(useTags)) {
2176 opts.useTags = Boolean.parseBoolean(cmd.substring(useTags.length()));
2177 continue;
2178 }
2179
2180 final String noOfTags = "--numoftags=";
2181 if (cmd.startsWith(noOfTags)) {
2182 opts.noOfTags = Integer.parseInt(cmd.substring(noOfTags.length()));
2183 continue;
2184 }
2185
2186 final String replicas = "--replicas=";
2187 if (cmd.startsWith(replicas)) {
2188 opts.replicas = Integer.parseInt(cmd.substring(replicas.length()));
2189 continue;
2190 }
2191
2192 final String filterOutAll = "--filterAll";
2193 if (cmd.startsWith(filterOutAll)) {
2194 opts.filterAll = true;
2195 continue;
2196 }
2197
2198 final String size = "--size=";
2199 if (cmd.startsWith(size)) {
2200 opts.size = Float.parseFloat(cmd.substring(size.length()));
2201 continue;
2202 }
2203
2204 final String splitPolicy = "--splitPolicy=";
2205 if (cmd.startsWith(splitPolicy)) {
2206 opts.splitPolicy = cmd.substring(splitPolicy.length());
2207 continue;
2208 }
2209
2210 final String randomSleep = "--randomSleep=";
2211 if (cmd.startsWith(randomSleep)) {
2212 opts.randomSleep = Integer.parseInt(cmd.substring(randomSleep.length()));
2213 continue;
2214 }
2215
2216 final String bloomFilter = "--bloomFilter=";
2217 if (cmd.startsWith(bloomFilter)) {
2218 opts.bloomType = BloomType.valueOf(cmd.substring(bloomFilter.length()));
2219 continue;
2220 }
2221
2222 final String blockSize = "--blockSize=";
2223 if(cmd.startsWith(blockSize) ) {
2224 opts.blockSize = Integer.parseInt(cmd.substring(blockSize.length()));
2225 }
2226
2227 final String valueSize = "--valueSize=";
2228 if (cmd.startsWith(valueSize)) {
2229 opts.valueSize = Integer.parseInt(cmd.substring(valueSize.length()));
2230 continue;
2231 }
2232
2233 final String valueRandom = "--valueRandom";
2234 if (cmd.startsWith(valueRandom)) {
2235 opts.valueRandom = true;
2236 if (opts.valueZipf) {
2237 throw new IllegalStateException("Either valueZipf or valueRandom but not both");
2238 }
2239 continue;
2240 }
2241
2242 final String valueZipf = "--valueZipf";
2243 if (cmd.startsWith(valueZipf)) {
2244 opts.valueZipf = true;
2245 if (opts.valueRandom) {
2246 throw new IllegalStateException("Either valueZipf or valueRandom but not both");
2247 }
2248 continue;
2249 }
2250
2251 final String period = "--period=";
2252 if (cmd.startsWith(period)) {
2253 opts.period = Integer.parseInt(cmd.substring(period.length()));
2254 continue;
2255 }
2256
2257 final String addColumns = "--addColumns=";
2258 if (cmd.startsWith(addColumns)) {
2259 opts.addColumns = Boolean.parseBoolean(cmd.substring(addColumns.length()));
2260 continue;
2261 }
2262
2263 final String columns = "--columns=";
2264 if (cmd.startsWith(columns)) {
2265 opts.columns = Integer.parseInt(cmd.substring(columns.length()));
2266 continue;
2267 }
2268
2269 final String families = "--families=";
2270 if (cmd.startsWith(families)) {
2271 opts.families = Integer.parseInt(cmd.substring(families.length()));
2272 continue;
2273 }
2274
2275 final String caching = "--caching=";
2276 if (cmd.startsWith(caching)) {
2277 opts.caching = Integer.parseInt(cmd.substring(caching.length()));
2278 continue;
2279 }
2280
2281 final String bufferSize = "--bufferSize=";
2282 if (cmd.startsWith(bufferSize)) {
2283 opts.bufferSize = Long.parseLong(cmd.substring(bufferSize.length()));
2284 continue;
2285 }
2286
2287 if (isCommandClass(cmd)) {
2288 opts.cmdName = cmd;
2289 try {
2290 opts.numClientThreads = Integer.parseInt(args.remove());
2291 } catch (NoSuchElementException | NumberFormatException e) {
2292 throw new IllegalArgumentException("Command " + cmd + " does not have threads number", e);
2293 }
2294 int rowsPerGB = getRowsPerGB(opts);
2295 if (opts.size != DEFAULT_OPTS.size &&
2296 opts.perClientRunRows != DEFAULT_OPTS.perClientRunRows) {
2297 throw new IllegalArgumentException(rows + " and " + size + " are mutually exclusive arguments.");
2298 }
2299 if (opts.size != DEFAULT_OPTS.size) {
2300
2301 opts.totalRows = (int) opts.size * rowsPerGB;
2302 opts.perClientRunRows = opts.totalRows / opts.numClientThreads;
2303 } else {
2304 opts.totalRows = opts.perClientRunRows * opts.numClientThreads;
2305 opts.size = opts.totalRows / rowsPerGB;
2306 }
2307 break;
2308 } else {
2309 printUsageAndExit("ERROR: Unrecognized option/command: " + cmd, -1);
2310 }
2311
2312
2313 System.err.println("Error: Wrong option or command: " + cmd);
2314 args.add(cmd);
2315 break;
2316 }
2317 return opts;
2318 }
2319
2320 static int getRowsPerGB(final TestOptions opts) {
2321 return ONE_GB / ((opts.valueRandom? opts.valueSize/2: opts.valueSize) * opts.getColumns() *
2322 opts.getFamilies());
2323 }
2324
2325 @Override
2326 public int run(String[] args) throws Exception {
2327
2328
2329 int errCode = -1;
2330 if (args.length < 1) {
2331 printUsage();
2332 return errCode;
2333 }
2334
2335 try {
2336 LinkedList<String> argv = new LinkedList<String>();
2337 argv.addAll(Arrays.asList(args));
2338 TestOptions opts = parseOpts(argv);
2339
2340
2341 if (!argv.isEmpty()) {
2342 errCode = 0;
2343 printUsage();
2344 return errCode;
2345 }
2346
2347
2348 if (opts.numClientThreads <= 0) {
2349 throw new IllegalArgumentException("Number of clients must be > 0");
2350 }
2351
2352
2353 if (opts.cmdName == null) {
2354 printUsage();
2355 return errCode;
2356 }
2357
2358 Class<? extends Test> cmdClass = determineCommandClass(opts.cmdName);
2359 if (cmdClass != null) {
2360 runTest(cmdClass, opts);
2361 errCode = 0;
2362 }
2363
2364 } catch (Exception e) {
2365 e.printStackTrace();
2366 }
2367
2368 return errCode;
2369 }
2370
2371 private static boolean isCommandClass(String cmd) {
2372 return COMMANDS.containsKey(cmd);
2373 }
2374
2375 private static Class<? extends Test> determineCommandClass(String cmd) {
2376 CmdDescriptor descriptor = COMMANDS.get(cmd);
2377 return descriptor != null ? descriptor.getCmdClass() : null;
2378 }
2379
2380 public static void main(final String[] args) throws Exception {
2381 int res = ToolRunner.run(new PerformanceEvaluation(HBaseConfiguration.create()), args);
2382 System.exit(res);
2383 }
2384 }