You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by al...@apache.org on 2018/05/24 11:38:06 UTC
hbase git commit: Revert "HBASE-20601 Add multiPut support and other
miscellaneous to PE"
Repository: hbase
Updated Branches:
refs/heads/master 320a3332e -> 4005a0c4d
Revert "HBASE-20601 Add multiPut support and other miscellaneous to PE"
Project: http://git-wip-us.apache.org/repos/asf/hbase/repo
Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/4005a0c4
Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/4005a0c4
Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/4005a0c4
Branch: refs/heads/master
Commit: 4005a0c4d368a9846f60026614a425192cc247ec
Parents: 320a333
Author: Allan Yang <al...@apache.org>
Authored: Thu May 24 19:37:21 2018 +0800
Committer: Allan Yang <al...@163.com>
Committed: Thu May 24 19:37:21 2018 +0800
----------------------------------------------------------------------
.../hadoop/hbase/PerformanceEvaluation.java | 344 ++++++++-----------
.../hadoop/hbase/TestPerformanceEvaluation.java | 46 ---
2 files changed, 140 insertions(+), 250 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hbase/blob/4005a0c4/hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/PerformanceEvaluation.java
----------------------------------------------------------------------
diff --git a/hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/PerformanceEvaluation.java b/hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/PerformanceEvaluation.java
index 26ab88d..42acb5c 100644
--- a/hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/PerformanceEvaluation.java
+++ b/hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/PerformanceEvaluation.java
@@ -438,17 +438,6 @@ public class PerformanceEvaluation extends Configured implements Tool {
return splits;
}
- static void setupConnectionCount(final TestOptions opts) {
- if (opts.oneCon) {
- opts.connCount = 1;
- } else {
- if (opts.connCount == -1) {
- // set to thread number if connCount is not set
- opts.connCount = opts.numClientThreads;
- }
- }
- }
-
/*
* Run all clients in this vm each to its own thread.
*/
@@ -461,23 +450,14 @@ public class PerformanceEvaluation extends Configured implements Tool {
RunResult[] results = new RunResult[opts.numClientThreads];
ExecutorService pool = Executors.newFixedThreadPool(opts.numClientThreads,
new ThreadFactoryBuilder().setNameFormat("TestClient-%s").build());
- setupConnectionCount(opts);
- final Connection[] cons = new Connection[opts.connCount];
- final AsyncConnection[] asyncCons = new AsyncConnection[opts.connCount];
- for (int i = 0; i < opts.connCount; i++) {
- cons[i] = ConnectionFactory.createConnection(conf);
- asyncCons[i] = ConnectionFactory.createAsyncConnection(conf).get();
- }
- LOG.info("Created " + opts.connCount + " connections for " +
- opts.numClientThreads + " threads");
+ final Connection con = ConnectionFactory.createConnection(conf);
+ final AsyncConnection asyncCon = ConnectionFactory.createAsyncConnection(conf).get();
for (int i = 0; i < threads.length; i++) {
final int index = i;
threads[i] = pool.submit(new Callable<RunResult>() {
@Override
public RunResult call() throws Exception {
TestOptions threadOpts = new TestOptions(opts);
- final Connection con = cons[index % cons.length];
- final AsyncConnection asyncCon = asyncCons[index % asyncCons.length];
if (threadOpts.startRow == 0) threadOpts.startRow = index * threadOpts.perClientRunRows;
RunResult run = runOneClient(cmd, conf, con, asyncCon, threadOpts, new Status() {
@Override
@@ -505,26 +485,16 @@ public class PerformanceEvaluation extends Configured implements Tool {
+ Arrays.toString(results));
Arrays.sort(results);
long total = 0;
- float avgLatency = 0 ;
- float avgTPS = 0;
for (RunResult result : results) {
total += result.duration;
- avgLatency += result.hist.getSnapshot().getMean();
- avgTPS += opts.perClientRunRows * 1.0f / result.duration;
}
- avgTPS *= 1000; // ms to second
- avgLatency = avgLatency / results.length;
- LOG.info("[" + test + " duration ]"
+ LOG.info("[" + test + "]"
+ "\tMin: " + results[0] + "ms"
+ "\tMax: " + results[results.length - 1] + "ms"
+ "\tAvg: " + (total / results.length) + "ms");
- LOG.info("[ Avg latency (us)]\t" + Math.round(avgLatency));
- LOG.info("[ Avg TPS/QPS]\t" + Math.round(avgTPS) + "\t row per second");
- for (int i = 0; i < opts.connCount; i++) {
- cons[i].close();
- asyncCons[i].close();
- }
+ con.close();
+ asyncCon.close();
return results;
}
@@ -678,12 +648,10 @@ public class PerformanceEvaluation extends Configured implements Tool {
boolean writeToWAL = true;
boolean autoFlush = false;
boolean oneCon = false;
- int connCount = -1; //wil decide the actual num later
boolean useTags = false;
int noOfTags = 1;
boolean reportLatency = false;
int multiGet = 0;
- int multiPut = 0;
int randomSleep = 0;
boolean inMemoryCF = false;
int presplitRegions = 0;
@@ -732,12 +700,10 @@ public class PerformanceEvaluation extends Configured implements Tool {
this.writeToWAL = that.writeToWAL;
this.autoFlush = that.autoFlush;
this.oneCon = that.oneCon;
- this.connCount = that.connCount;
this.useTags = that.useTags;
this.noOfTags = that.noOfTags;
this.reportLatency = that.reportLatency;
this.multiGet = that.multiGet;
- this.multiPut = that.multiPut;
this.inMemoryCF = that.inMemoryCF;
this.presplitRegions = that.presplitRegions;
this.replicas = that.replicas;
@@ -892,14 +858,6 @@ public class PerformanceEvaluation extends Configured implements Tool {
this.oneCon = oneCon;
}
- public int getConnCount() {
- return connCount;
- }
-
- public void setConnCount(int connCount) {
- this.connCount = connCount;
- }
-
public void setUseTags(boolean useTags) {
this.useTags = useTags;
}
@@ -916,10 +874,6 @@ public class PerformanceEvaluation extends Configured implements Tool {
this.multiGet = multiGet;
}
- public void setMultiPut(int multiPut) {
- this.multiPut = multiPut;
- }
-
public void setInMemoryCF(boolean inMemoryCF) {
this.inMemoryCF = inMemoryCF;
}
@@ -1028,10 +982,6 @@ public class PerformanceEvaluation extends Configured implements Tool {
return multiGet;
}
- public int getMultiPut() {
- return multiPut;
- }
-
public boolean isInMemoryCF() {
return inMemoryCF;
}
@@ -1248,9 +1198,12 @@ public class PerformanceEvaluation extends Configured implements Tool {
bytesInResultsHistogram = YammerHistogramUtils.newHistogram(new UniformReservoir(1024 * 500));
bytesInRemoteResultsHistogram = YammerHistogramUtils.newHistogram(new UniformReservoir(1024 * 500));
+ createConnection();
onStartup();
}
+ abstract void createConnection() throws IOException;
+
abstract void onStartup() throws IOException;
void testTakedown() throws IOException {
@@ -1264,14 +1217,10 @@ public class PerformanceEvaluation extends Configured implements Tool {
latencyHistogram));
status.setStatus("Num measures (latency) : " + latencyHistogram.getCount());
status.setStatus(YammerHistogramUtils.getPrettyHistogramReport(latencyHistogram));
- if (valueSizeHistogram.getCount() > 0) {
- status.setStatus("ValueSize (bytes) : "
- + YammerHistogramUtils.getHistogramReport(valueSizeHistogram));
- status.setStatus("Num measures (ValueSize): " + valueSizeHistogram.getCount());
- status.setStatus(YammerHistogramUtils.getPrettyHistogramReport(valueSizeHistogram));
- } else {
- status.setStatus("No valueSize statistics available");
- }
+ status.setStatus("ValueSize (bytes) : "
+ + YammerHistogramUtils.getHistogramReport(valueSizeHistogram));
+ status.setStatus("Num measures (ValueSize): " + valueSizeHistogram.getCount());
+ status.setStatus(YammerHistogramUtils.getPrettyHistogramReport(valueSizeHistogram));
if (rpcCallsHistogram.getCount() > 0) {
status.setStatus("rpcCalls (count): " +
YammerHistogramUtils.getHistogramReport(rpcCallsHistogram));
@@ -1297,11 +1246,13 @@ public class PerformanceEvaluation extends Configured implements Tool {
YammerHistogramUtils.getHistogramReport(bytesInRemoteResultsHistogram));
}
}
+ closeConnection();
receiverHost.closeReceivers();
}
abstract void onTakedown() throws IOException;
+ abstract void closeConnection() throws IOException;
/*
* Run test
@@ -1341,16 +1292,14 @@ public class PerformanceEvaluation extends Configured implements Tool {
for (int i = startRow; i < lastRow; i++) {
if (i % everyN != 0) continue;
long startTime = System.nanoTime();
- boolean requestSent = false;
- try (TraceScope scope = TraceUtil.createTrace("test row");){
- requestSent = testRow(i);
+ try (TraceScope scope = TraceUtil.createTrace("test row")){
+ testRow(i);
}
if ( (i - startRow) > opts.measureAfter) {
- // If multiget or multiput is enabled, say set to 10, testRow() returns immediately
- // first 9 times and sends the actual get request in the 10th iteration.
- // We should only set latency when actual request is sent because otherwise
- // it turns out to be 0.
- if (requestSent) {
+ // If multiget is enabled, say set to 10, testRow() returns immediately first 9 times
+ // and sends the actual get request in the 10th iteration. We should only set latency
+ // when actual request is sent because otherwise it turns out to be 0.
+ if (opts.multiGet == 0 || (i - startRow + 1) % opts.multiGet == 0) {
latencyHistogram.update((System.nanoTime() - startTime) / 1000);
}
if (status != null && i > 0 && (i % getReportingPeriod()) == 0) {
@@ -1375,15 +1324,11 @@ public class PerformanceEvaluation extends Configured implements Tool {
return YammerHistogramUtils.getShortHistogramReport(this.valueSizeHistogram);
}
-
- /**
- * Test for individual row.
- * @param i Row index.
- * @return true if the row was sent to server and need to record metrics.
- * False if not, multiGet and multiPut e.g., the rows are sent
- * to server only if enough gets/puts are gathered.
- */
- abstract boolean testRow(final int i) throws IOException, InterruptedException;
+ /*
+ * Test for individual row.
+ * @param i Row index.
+ */
+ abstract void testRow(final int i) throws IOException, InterruptedException;
}
static abstract class Test extends TestBase {
@@ -1393,6 +1338,20 @@ public class PerformanceEvaluation extends Configured implements Tool {
super(con == null ? HBaseConfiguration.create() : con.getConfiguration(), options, status);
this.connection = con;
}
+
+ @Override
+ void createConnection() throws IOException {
+ if (!opts.isOneCon()) {
+ this.connection = ConnectionFactory.createConnection(conf);
+ }
+ }
+
+ @Override
+ void closeConnection() throws IOException {
+ if (!opts.isOneCon()) {
+ this.connection.close();
+ }
+ }
}
static abstract class AsyncTest extends TestBase {
@@ -1402,6 +1361,24 @@ public class PerformanceEvaluation extends Configured implements Tool {
super(con == null ? HBaseConfiguration.create() : con.getConfiguration(), options, status);
this.connection = con;
}
+
+ @Override
+ void createConnection() {
+ if (!opts.isOneCon()) {
+ try {
+ this.connection = ConnectionFactory.createAsyncConnection(conf).get();
+ } catch (InterruptedException | ExecutionException e) {
+ LOG.error("Failed to create async connection", e);
+ }
+ }
+ }
+
+ @Override
+ void closeConnection() throws IOException {
+ if (!opts.isOneCon()) {
+ this.connection.close();
+ }
+ }
}
static abstract class TableTest extends Test {
@@ -1454,7 +1431,7 @@ public class PerformanceEvaluation extends Configured implements Tool {
}
@Override
- boolean testRow(final int i) throws IOException, InterruptedException {
+ void testRow(final int i) throws IOException, InterruptedException {
if (opts.randomSleep > 0) {
Thread.sleep(rd.nextInt(opts.randomSleep));
}
@@ -1483,8 +1460,6 @@ public class PerformanceEvaluation extends Configured implements Tool {
this.table.get(this.gets).stream().map(f -> propagate(f::get)).toArray(Result[]::new);
updateValueSize(rs);
this.gets.clear();
- } else {
- return false;
}
} else {
updateValueSize(this.table.get(get).get());
@@ -1492,7 +1467,6 @@ public class PerformanceEvaluation extends Configured implements Tool {
} catch (ExecutionException e) {
throw new IOException(e);
}
- return true;
}
public static RuntimeException runtime(Throwable e) {
@@ -1526,15 +1500,43 @@ public class PerformanceEvaluation extends Configured implements Tool {
}
}
- static class AsyncRandomWriteTest extends AsyncSequentialWriteTest {
-
+ static class AsyncRandomWriteTest extends AsyncTableTest {
AsyncRandomWriteTest(AsyncConnection con, TestOptions options, Status status) {
super(con, options, status);
}
@Override
- protected byte[] generateRow(final int i) {
- return getRandomRow(this.rand, opts.totalRows);
+ void testRow(final int i) throws IOException, InterruptedException {
+ byte[] row = getRandomRow(this.rand, opts.totalRows);
+ Put put = new Put(row);
+ for (int family = 0; family < opts.families; family++) {
+ byte[] familyName = Bytes.toBytes(FAMILY_NAME_BASE + family);
+ for (int column = 0; column < opts.columns; column++) {
+ byte[] qualifier = column == 0 ? COLUMN_ZERO : Bytes.toBytes("" + column);
+ byte[] value = generateData(this.rand, getValueLength(this.rand));
+ if (opts.useTags) {
+ byte[] tag = generateData(this.rand, TAG_LENGTH);
+ Tag[] tags = new Tag[opts.noOfTags];
+ for (int n = 0; n < opts.noOfTags; n++) {
+ Tag t = new ArrayBackedTag((byte) n, tag);
+ tags[n] = t;
+ }
+ KeyValue kv =
+ new KeyValue(row, familyName, qualifier, HConstants.LATEST_TIMESTAMP, value, tags);
+ put.add(kv);
+ updateValueSize(kv.getValueLength());
+ } else {
+ put.addColumn(familyName, qualifier, value);
+ updateValueSize(value.length);
+ }
+ }
+ }
+ put.setDurability(opts.writeToWAL ? Durability.SYNC_WAL : Durability.SKIP_WAL);
+ try {
+ table.put(put).get();
+ } catch (ExecutionException e) {
+ throw new IOException(e);
+ }
}
}
@@ -1563,7 +1565,7 @@ public class PerformanceEvaluation extends Configured implements Tool {
}
@Override
- boolean testRow(final int i) throws IOException {
+ void testRow(final int i) throws IOException {
if (this.testScanner == null) {
Scan scan =
new Scan().withStartRow(format(opts.startRow)).setCaching(opts.caching)
@@ -1587,7 +1589,6 @@ public class PerformanceEvaluation extends Configured implements Tool {
}
Result r = testScanner.next();
updateValueSize(r);
- return true;
}
}
@@ -1597,7 +1598,7 @@ public class PerformanceEvaluation extends Configured implements Tool {
}
@Override
- boolean testRow(final int i) throws IOException, InterruptedException {
+ void testRow(final int i) throws IOException, InterruptedException {
Get get = new Get(format(i));
for (int family = 0; family < opts.families; family++) {
byte[] familyName = Bytes.toBytes(FAMILY_NAME_BASE + family);
@@ -1618,29 +1619,17 @@ public class PerformanceEvaluation extends Configured implements Tool {
} catch (ExecutionException e) {
throw new IOException(e);
}
- return true;
}
}
static class AsyncSequentialWriteTest extends AsyncTableTest {
- private ArrayList<Put> puts;
-
AsyncSequentialWriteTest(AsyncConnection con, TestOptions options, Status status) {
super(con, options, status);
- if (opts.multiPut > 0) {
- LOG.info("MultiPut enabled. Sending PUTs in batches of " + opts.multiPut + ".");
- this.puts = new ArrayList<>(opts.multiPut);
- }
- }
-
- protected byte[] generateRow(final int i) {
- return format(i);
}
@Override
- @SuppressWarnings("ReturnValueIgnored")
- boolean testRow(final int i) throws IOException, InterruptedException {
- byte[] row = generateRow(i);
+ void testRow(final int i) throws IOException, InterruptedException {
+ byte[] row = format(i);
Put put = new Put(row);
for (int family = 0; family < opts.families; family++) {
byte[] familyName = Bytes.toBytes(FAMILY_NAME_BASE + family);
@@ -1667,21 +1656,9 @@ public class PerformanceEvaluation extends Configured implements Tool {
put.setDurability(opts.writeToWAL ? Durability.SYNC_WAL : Durability.SKIP_WAL);
try {
table.put(put).get();
- if (opts.multiPut > 0) {
- this.puts.add(put);
- if (this.puts.size() == opts.multiPut) {
- this.table.put(puts).stream().map(f -> AsyncRandomReadTest.propagate(f::get));
- this.puts.clear();
- } else {
- return false;
- }
- } else {
- table.put(put).get();
- }
} catch (ExecutionException e) {
throw new IOException(e);
}
- return true;
}
}
@@ -1714,7 +1691,7 @@ public class PerformanceEvaluation extends Configured implements Tool {
}
@Override
- boolean testRow(final int i) throws IOException {
+ void testRow(final int i) throws IOException {
Scan scan = new Scan().withStartRow(getRandomRow(this.rand, opts.totalRows))
.setCaching(opts.caching).setCacheBlocks(opts.cacheBlocks)
.setAsyncPrefetch(opts.asyncPrefetch).setReadType(opts.scanReadType)
@@ -1745,7 +1722,6 @@ public class PerformanceEvaluation extends Configured implements Tool {
updateScanMetrics(s.getScanMetrics());
s.close();
}
- return true;
}
@Override
@@ -1762,7 +1738,7 @@ public class PerformanceEvaluation extends Configured implements Tool {
}
@Override
- boolean testRow(final int i) throws IOException {
+ void testRow(final int i) throws IOException {
Pair<byte[], byte[]> startAndStopRow = getStartAndStopRow();
Scan scan = new Scan().withStartRow(startAndStopRow.getFirst())
.withStopRow(startAndStopRow.getSecond()).setCaching(opts.caching)
@@ -1799,7 +1775,6 @@ public class PerformanceEvaluation extends Configured implements Tool {
updateScanMetrics(s.getScanMetrics());
s.close();
}
- return true;
}
protected abstract Pair<byte[],byte[]> getStartAndStopRow();
@@ -1876,7 +1851,7 @@ public class PerformanceEvaluation extends Configured implements Tool {
}
@Override
- boolean testRow(final int i) throws IOException, InterruptedException {
+ void testRow(final int i) throws IOException, InterruptedException {
if (opts.randomSleep > 0) {
Thread.sleep(rd.nextInt(opts.randomSleep));
}
@@ -1903,13 +1878,10 @@ public class PerformanceEvaluation extends Configured implements Tool {
Result [] rs = this.table.get(this.gets);
updateValueSize(rs);
this.gets.clear();
- } else {
- return false;
}
} else {
updateValueSize(this.table.get(get));
}
- return true;
}
@Override
@@ -1928,17 +1900,44 @@ public class PerformanceEvaluation extends Configured implements Tool {
}
}
- static class RandomWriteTest extends SequentialWriteTest {
+ static class RandomWriteTest extends BufferedMutatorTest {
RandomWriteTest(Connection con, TestOptions options, Status status) {
super(con, options, status);
}
@Override
- protected byte[] generateRow(final int i) {
- return getRandomRow(this.rand, opts.totalRows);
+ void testRow(final int i) throws IOException {
+ byte[] row = getRandomRow(this.rand, opts.totalRows);
+ Put put = new Put(row);
+ for (int family = 0; family < opts.families; family++) {
+ byte[] familyName = Bytes.toBytes(FAMILY_NAME_BASE + family);
+ for (int column = 0; column < opts.columns; column++) {
+ byte [] qualifier = column == 0? COLUMN_ZERO: Bytes.toBytes("" + column);
+ byte[] value = generateData(this.rand, getValueLength(this.rand));
+ if (opts.useTags) {
+ byte[] tag = generateData(this.rand, TAG_LENGTH);
+ Tag[] tags = new Tag[opts.noOfTags];
+ for (int n = 0; n < opts.noOfTags; n++) {
+ Tag t = new ArrayBackedTag((byte) n, tag);
+ tags[n] = t;
+ }
+ KeyValue kv = new KeyValue(row, familyName, qualifier, HConstants.LATEST_TIMESTAMP,
+ value, tags);
+ put.add(kv);
+ updateValueSize(kv.getValueLength());
+ } else {
+ put.addColumn(familyName, qualifier, value);
+ updateValueSize(value.length);
+ }
+ }
+ }
+ put.setDurability(opts.writeToWAL ? Durability.SYNC_WAL : Durability.SKIP_WAL);
+ if (opts.autoFlush) {
+ table.put(put);
+ } else {
+ mutator.mutate(put);
+ }
}
-
-
}
static class ScanTest extends TableTest {
@@ -1958,7 +1957,7 @@ public class PerformanceEvaluation extends Configured implements Tool {
@Override
- boolean testRow(final int i) throws IOException {
+ void testRow(final int i) throws IOException {
if (this.testScanner == null) {
Scan scan = new Scan().withStartRow(format(opts.startRow)).setCaching(opts.caching)
.setCacheBlocks(opts.cacheBlocks).setAsyncPrefetch(opts.asyncPrefetch)
@@ -1981,7 +1980,6 @@ public class PerformanceEvaluation extends Configured implements Tool {
}
Result r = testScanner.next();
updateValueSize(r);
- return true;
}
}
@@ -2021,7 +2019,7 @@ public class PerformanceEvaluation extends Configured implements Tool {
}
@Override
- boolean testRow(final int i) throws IOException {
+ void testRow(final int i) throws IOException {
Increment increment = new Increment(format(i));
// unlike checkAndXXX tests, which make most sense to do on a single value,
// if multiple families are specified for an increment test we assume it is
@@ -2031,7 +2029,6 @@ public class PerformanceEvaluation extends Configured implements Tool {
increment.addColumn(familyName, getQualifier(), 1l);
}
updateValueSize(this.table.increment(increment));
- return true;
}
}
@@ -2041,7 +2038,7 @@ public class PerformanceEvaluation extends Configured implements Tool {
}
@Override
- boolean testRow(final int i) throws IOException {
+ void testRow(final int i) throws IOException {
byte [] bytes = format(i);
Append append = new Append(bytes);
// unlike checkAndXXX tests, which make most sense to do on a single value,
@@ -2052,7 +2049,6 @@ public class PerformanceEvaluation extends Configured implements Tool {
append.addColumn(familyName, getQualifier(), bytes);
}
updateValueSize(this.table.append(append));
- return true;
}
}
@@ -2062,7 +2058,7 @@ public class PerformanceEvaluation extends Configured implements Tool {
}
@Override
- boolean testRow(final int i) throws IOException {
+ void testRow(final int i) throws IOException {
final byte [] bytes = format(i);
// checkAndXXX tests operate on only a single value
// Put a known value so when we go to check it, it is there.
@@ -2073,7 +2069,6 @@ public class PerformanceEvaluation extends Configured implements Tool {
mutations.add(put);
this.table.checkAndMutate(bytes, FAMILY_ZERO).qualifier(getQualifier())
.ifEquals(bytes).thenMutate(mutations);
- return true;
}
}
@@ -2083,7 +2078,7 @@ public class PerformanceEvaluation extends Configured implements Tool {
}
@Override
- boolean testRow(final int i) throws IOException {
+ void testRow(final int i) throws IOException {
final byte [] bytes = format(i);
// checkAndXXX tests operate on only a single value
// Put a known value so when we go to check it, it is there.
@@ -2092,7 +2087,6 @@ public class PerformanceEvaluation extends Configured implements Tool {
this.table.put(put);
this.table.checkAndMutate(bytes, FAMILY_ZERO).qualifier(getQualifier())
.ifEquals(bytes).thenPut(put);
- return true;
}
}
@@ -2102,7 +2096,7 @@ public class PerformanceEvaluation extends Configured implements Tool {
}
@Override
- boolean testRow(final int i) throws IOException {
+ void testRow(final int i) throws IOException {
final byte [] bytes = format(i);
// checkAndXXX tests operate on only a single value
// Put a known value so when we go to check it, it is there.
@@ -2113,7 +2107,6 @@ public class PerformanceEvaluation extends Configured implements Tool {
delete.addColumn(FAMILY_ZERO, getQualifier());
this.table.checkAndMutate(bytes, FAMILY_ZERO).qualifier(getQualifier())
.ifEquals(bytes).thenDelete(delete);
- return true;
}
}
@@ -2123,7 +2116,7 @@ public class PerformanceEvaluation extends Configured implements Tool {
}
@Override
- boolean testRow(final int i) throws IOException {
+ void testRow(final int i) throws IOException {
Get get = new Get(format(i));
for (int family = 0; family < opts.families; family++) {
byte[] familyName = Bytes.toBytes(FAMILY_NAME_BASE + family);
@@ -2140,29 +2133,17 @@ public class PerformanceEvaluation extends Configured implements Tool {
get.setFilter(new FilterAllFilter());
}
updateValueSize(table.get(get));
- return true;
}
}
static class SequentialWriteTest extends BufferedMutatorTest {
- private ArrayList<Put> puts;
-
-
SequentialWriteTest(Connection con, TestOptions options, Status status) {
super(con, options, status);
- if (opts.multiPut > 0) {
- LOG.info("MultiPut enabled. Sending PUTs in batches of " + opts.multiPut + ".");
- this.puts = new ArrayList<>(opts.multiPut);
- }
- }
-
- protected byte[] generateRow(final int i) {
- return format(i);
}
@Override
- boolean testRow(final int i) throws IOException {
- byte[] row = generateRow(i);
+ void testRow(final int i) throws IOException {
+ byte[] row = format(i);
Put put = new Put(row);
for (int family = 0; family < opts.families; family++) {
byte familyName[] = Bytes.toBytes(FAMILY_NAME_BASE + family);
@@ -2188,21 +2169,10 @@ public class PerformanceEvaluation extends Configured implements Tool {
}
put.setDurability(opts.writeToWAL ? Durability.SYNC_WAL : Durability.SKIP_WAL);
if (opts.autoFlush) {
- if (opts.multiPut > 0) {
- this.puts.add(put);
- if (this.puts.size() == opts.multiPut) {
- table.put(this.puts);
- this.puts.clear();
- } else {
- return false;
- }
- } else {
- table.put(put);
- }
+ table.put(put);
} else {
mutator.mutate(put);
}
- return true;
}
}
@@ -2214,7 +2184,7 @@ public class PerformanceEvaluation extends Configured implements Tool {
}
@Override
- boolean testRow(int i) throws IOException {
+ void testRow(int i) throws IOException {
byte[] value = generateData(this.rand, getValueLength(this.rand));
Scan scan = constructScan(value);
ResultScanner scanner = null;
@@ -2229,7 +2199,6 @@ public class PerformanceEvaluation extends Configured implements Tool {
scanner.close();
}
}
- return true;
}
protected Scan constructScan(byte[] valuePrefix) throws IOException {
@@ -2411,11 +2380,6 @@ public class PerformanceEvaluation extends Configured implements Tool {
System.err.println(" nomapred Run multiple clients using threads " +
"(rather than use mapreduce)");
System.err.println(" oneCon all the threads share the same connection. Default: False");
- System.err.println(" connCount connections all threads share. "
- + "For example, if set to 2, then all thread share 2 connection. "
- + "Default: depend on oneCon parameter. if oneCon set to true, then connCount=1, "
- + "if not, connCount=thread number");
-
System.err.println(" sampleRate Execute test on a sample of total " +
"rows. Only supported by randomRead. Default: 1.0");
System.err.println(" period Report every 'period' rows: " +
@@ -2452,8 +2416,6 @@ public class PerformanceEvaluation extends Configured implements Tool {
"'valueSize' in zipf form: Default: Not set.");
System.err.println(" writeToWAL Set writeToWAL on puts. Default: True");
System.err.println(" autoFlush Set autoFlush on htable. Default: False");
- System.err.println(" multiPut Batch puts together into groups of N. Only supported " +
- "by write. If multiPut is bigger than 0, autoFlush need to set to true. Default: 0");
System.err.println(" presplit Create presplit table. If a table with same name exists,"
+ " it'll be deleted and recreated (instead of verifying count of its existing regions). "
+ "Recommended for accurate perf analysis (see guide). Default: disabled");
@@ -2606,29 +2568,12 @@ public class PerformanceEvaluation extends Configured implements Tool {
final String autoFlush = "--autoFlush=";
if (cmd.startsWith(autoFlush)) {
opts.autoFlush = Boolean.parseBoolean(cmd.substring(autoFlush.length()));
- if (!opts.autoFlush && opts.multiPut > 0) {
- throw new IllegalArgumentException("autoFlush must be true when multiPut is more than 0");
- }
continue;
}
final String onceCon = "--oneCon=";
if (cmd.startsWith(onceCon)) {
opts.oneCon = Boolean.parseBoolean(cmd.substring(onceCon.length()));
- if (opts.oneCon && opts.connCount > 1) {
- throw new IllegalArgumentException("oneCon is set to true, "
- + "connCount should not bigger than 1");
- }
- continue;
- }
-
- final String connCount = "--connCount=";
- if (cmd.startsWith(connCount)) {
- opts.connCount = Integer.parseInt(cmd.substring(connCount.length()));
- if (opts.oneCon && opts.connCount > 1) {
- throw new IllegalArgumentException("oneCon is set to true, "
- + "connCount should not bigger than 1");
- }
continue;
}
@@ -2644,15 +2589,6 @@ public class PerformanceEvaluation extends Configured implements Tool {
continue;
}
- final String multiPut = "--multiPut=";
- if (cmd.startsWith(multiPut)) {
- opts.multiPut = Integer.parseInt(cmd.substring(multiPut.length()));
- if (!opts.autoFlush && opts.multiPut > 0) {
- throw new IllegalArgumentException("autoFlush must be true when multiPut is more than 0");
- }
- continue;
- }
-
final String useTags = "--usetags=";
if (cmd.startsWith(useTags)) {
opts.useTags = Boolean.parseBoolean(cmd.substring(useTags.length()));
http://git-wip-us.apache.org/repos/asf/hbase/blob/4005a0c4/hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/TestPerformanceEvaluation.java
----------------------------------------------------------------------
diff --git a/hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/TestPerformanceEvaluation.java b/hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/TestPerformanceEvaluation.java
index 2e1ecd0..b2a89c8 100644
--- a/hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/TestPerformanceEvaluation.java
+++ b/hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/TestPerformanceEvaluation.java
@@ -243,50 +243,4 @@ public class TestPerformanceEvaluation {
assertTrue(e.getCause() instanceof NoSuchElementException);
}
}
-
- @Test
- public void testParseOptsMultiPuts() {
- Queue<String> opts = new LinkedList<>();
- String cmdName = "sequentialWrite";
- opts.offer("--multiPut=10");
- opts.offer(cmdName);
- opts.offer("64");
- PerformanceEvaluation.TestOptions options = null;
- try {
- options = PerformanceEvaluation.parseOpts(opts);
- fail("should fail");
- } catch (IllegalArgumentException e) {
- System.out.println(e.getMessage());
- }
- ((LinkedList<String>) opts).offerFirst("--multiPut=10");
- ((LinkedList<String>) opts).offerFirst("--autoFlush=true");
- options = PerformanceEvaluation.parseOpts(opts);
- assertNotNull(options);
- assertNotNull(options.getCmdName());
- assertEquals(cmdName, options.getCmdName());
- assertTrue(options.getMultiPut() == 10);
- }
-
- @Test
- public void testParseOptsConnCount() {
- Queue<String> opts = new LinkedList<>();
- String cmdName = "sequentialWrite";
- opts.offer("--oneCon=true");
- opts.offer("--connCount=10");
- opts.offer(cmdName);
- opts.offer("64");
- PerformanceEvaluation.TestOptions options = null;
- try {
- options = PerformanceEvaluation.parseOpts(opts);
- fail("should fail");
- } catch (IllegalArgumentException e) {
- System.out.println(e.getMessage());
- }
- ((LinkedList<String>) opts).offerFirst("--connCount=10");
- options = PerformanceEvaluation.parseOpts(opts);
- assertNotNull(options);
- assertNotNull(options.getCmdName());
- assertEquals(cmdName, options.getCmdName());
- assertTrue(options.getConnCount() == 10);
- }
}