Adding a shell command to load test data into Blur.
Project: http://git-wip-us.apache.org/repos/asf/incubator-blur/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-blur/commit/89200000 Tree: http://git-wip-us.apache.org/repos/asf/incubator-blur/tree/89200000 Diff: http://git-wip-us.apache.org/repos/asf/incubator-blur/diff/89200000 Branch: refs/heads/master Commit: 8920000083335b5c7675acb37cb4173d0ec367d4 Parents: 728f4a8 Author: Aaron McCurry <[email protected]> Authored: Wed Aug 7 21:28:33 2013 -0400 Committer: Aaron McCurry <[email protected]> Committed: Wed Aug 7 21:28:33 2013 -0400 ---------------------------------------------------------------------- .../apache/blur/shell/LoadTestDataCommand.java | 60 ++++++++++++++++ .../main/java/org/apache/blur/shell/Main.java | 41 +++++------ .../org/apache/blur/thrift/util/LoadData.java | 73 +++++++++++++++----- 3 files changed, 138 insertions(+), 36 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/89200000/blur-shell/src/main/java/org/apache/blur/shell/LoadTestDataCommand.java ---------------------------------------------------------------------- diff --git a/blur-shell/src/main/java/org/apache/blur/shell/LoadTestDataCommand.java b/blur-shell/src/main/java/org/apache/blur/shell/LoadTestDataCommand.java new file mode 100644 index 0000000..f25a767 --- /dev/null +++ b/blur-shell/src/main/java/org/apache/blur/shell/LoadTestDataCommand.java @@ -0,0 +1,60 @@ +package org.apache.blur.shell; + +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +import java.io.IOException; +import java.io.PrintWriter; + +import org.apache.blur.thirdparty.thrift_0_9_0.TException; +import org.apache.blur.thrift.generated.Blur; +import org.apache.blur.thrift.generated.BlurException; +import org.apache.blur.thrift.util.LoadData; + +public class LoadTestDataCommand extends Command { + + @Override + public void doit(PrintWriter out, Blur.Iface client, String[] args) throws CommandException, TException, + BlurException { + if (args.length != 9) { + throw new CommandException("Invalid args: " + help()); + } + int c = 1; + String table = args[c++]; + boolean wal = Boolean.parseBoolean(args[c++]); + int numberRows = Integer.parseInt(args[c++]); + int numberRecordsPerRow = Integer.parseInt(args[c++]); + int numberOfFamilies = Integer.parseInt(args[c++]); + int numberOfColumns = Integer.parseInt(args[c++]); + int numberOfWords = Integer.parseInt(args[c++]); + int batch = Integer.parseInt(args[c++]); + try { + LoadData.runLoad(client, table, wal, numberRows, numberRecordsPerRow, numberOfFamilies, numberOfColumns, + numberOfWords, batch, out); + } catch (IOException e) { + out.println("Error " + e.getMessage()); + if (Main.debug) { + e.printStackTrace(); + } + } + } + + @Override + public String help() { + return "load test data, args; tablename wal(boolean) rows(int) recordsPerRow(int) families(int) columnsPerRecord(int) wordsPerColumn(int) batchSize(int)"; + } +} http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/89200000/blur-shell/src/main/java/org/apache/blur/shell/Main.java ---------------------------------------------------------------------- diff --git a/blur-shell/src/main/java/org/apache/blur/shell/Main.java b/blur-shell/src/main/java/org/apache/blur/shell/Main.java index 1a57088..acddd3c 100644 --- a/blur-shell/src/main/java/org/apache/blur/shell/Main.java +++ b/blur-shell/src/main/java/org/apache/blur/shell/Main.java @@ -134,25 +134,25 @@ public class Main { throw new CommandException("Invalid args: " + help()); } String clusterNamePassed = args[1]; - if(validateClusterName(client, clusterNamePassed)) { - cluster = clusterNamePassed; - out.println("cluster is now " + cluster); - }else{ - out.println("[ " + clusterNamePassed + " ]"+" is not a valid cluster name."); + if (validateClusterName(client, clusterNamePassed)) { + cluster = clusterNamePassed; + out.println("cluster is now " + cluster); + } else { + out.println("[ " + clusterNamePassed + " ]" + " is not a valid cluster name."); } } private boolean validateClusterName(Iface client, String clusterName) throws BlurException, TException { - List<String> clusterNamesList = client.shardClusterList(); - if(clusterNamesList != null && !clusterNamesList.isEmpty()){ - if(clusterNamesList.contains(clusterName)){ - return true; - } - } - return false; - } - - @Override + List<String> clusterNamesList = client.shardClusterList(); + if (clusterNamesList != null && !clusterNamesList.isEmpty()) { + if (clusterNamesList.contains(clusterName)) { + return true; + } + } + return false; + } + + @Override public String help() { return "set the cluster in use, args; clustername"; } @@ -346,6 +346,7 @@ public class Main { builder.put("safemodewait", new WaitInSafemodeCommand()); builder.put("top", new TopCommand()); builder.put("parse", new ParseCommand()); + builder.put("loadtestdata", new LoadTestDataCommand()); commands = builder.build(); CliShellOptions cliShellOptions = getCliShellOptions(args); @@ -403,9 +404,9 @@ public class Main { if (debug) { e.printStackTrace(out); } - }catch (BadConnectionException e){ - out.println(e.getMessage()); - }finally { + } catch (BadConnectionException e) { + out.println(e.getMessage()); + } finally { if (timed) { out.println("Last command took " + TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start) + "ms"); } @@ -425,8 +426,8 @@ public class Main { } catch (BlurException e) { out.println(e.getMessage()); e.printStackTrace(out); - } catch (BadConnectionException e){ - out.println(e.getMessage()); + } catch (BadConnectionException e) { + out.println(e.getMessage()); } out.close(); return; http://git-wip-us.apache.org/repos/asf/incubator-blur/blob/89200000/blur-thrift/src/main/java/org/apache/blur/thrift/util/LoadData.java ---------------------------------------------------------------------- diff --git a/blur-thrift/src/main/java/org/apache/blur/thrift/util/LoadData.java b/blur-thrift/src/main/java/org/apache/blur/thrift/util/LoadData.java index 93ddce6..cf5f62a 100644 --- a/blur-thrift/src/main/java/org/apache/blur/thrift/util/LoadData.java +++ b/blur-thrift/src/main/java/org/apache/blur/thrift/util/LoadData.java @@ -20,12 +20,15 @@ import java.io.BufferedReader; import java.io.IOException; import java.io.InputStream; import java.io.InputStreamReader; +import java.io.PrintWriter; import java.util.ArrayList; import java.util.List; import java.util.Random; +import java.util.concurrent.TimeUnit; import org.apache.blur.thirdparty.thrift_0_9_0.TException; import org.apache.blur.thrift.BlurClient; +import org.apache.blur.thrift.generated.Blur.Iface; import org.apache.blur.thrift.generated.BlurException; import org.apache.blur.thrift.generated.Column; import org.apache.blur.thrift.generated.Record; @@ -33,35 +36,51 @@ import org.apache.blur.thrift.generated.RecordMutation; import org.apache.blur.thrift.generated.RecordMutationType; import org.apache.blur.thrift.generated.RowMutation; import org.apache.blur.thrift.generated.RowMutationType; -import org.apache.blur.thrift.generated.Blur.Iface; - public class LoadData { + private static final long _5_SECONDS = TimeUnit.SECONDS.toNanos(5); private static Random random = new Random(); private static List<String> words = new ArrayList<String>(); public static void main(String[] args) throws BlurException, TException, IOException { - loadWords(); final boolean wal = true; final int numberOfColumns = 3; int numberRows = 100000; final int numberRecordsPerRow = 3; final int numberOfFamilies = 3; final int numberOfWords = 30; - int count = 0; - int max = 100; - long start = System.currentTimeMillis(); - final String table = args[1]; + int batch = 1; + String connectionString = args[0]; + String table = args[1]; + Iface client = BlurClient.getClient(connectionString); + runLoad(client, table, wal, numberRows, numberRecordsPerRow, numberOfFamilies, numberOfColumns, numberOfWords, + batch, new PrintWriter(System.out)); + } + + public static void runLoad(Iface client, String table, boolean wal, int numberRows, int numberRecordsPerRow, + int numberOfFamilies, int numberOfColumns, int numberOfWords, int batch, PrintWriter out) throws IOException, + BlurException, TException { + loadWords(); + + int countRow = 0; + int countRecord = 0; + final long start = System.currentTimeMillis(); + long s = start; + + List<RowMutation> mutations = new ArrayList<RowMutation>(); + + long ts = System.nanoTime(); for (int i = 0; i < numberRows; i++) { - if (count >= max) { - double seconds = (System.currentTimeMillis() - start) / 1000.0; - double rate = i / seconds; - System.out.println("Rows indexed [" + i + "] at [" + rate + "/s]"); - count = 0; + long now = System.nanoTime(); + if (ts + _5_SECONDS < now) { + printPerformance(out, countRow, countRecord, start, s, i); + countRow = 0; + countRecord = 0; + s = System.currentTimeMillis(); + ts = System.nanoTime(); } - Iface client = BlurClient.getClient(args[0]); RowMutation mutation = new RowMutation(); mutation.setTable(table); String rowId = getRowId(); @@ -70,10 +89,32 @@ public class LoadData { mutation.setRowMutationType(RowMutationType.REPLACE_ROW); for (int j = 0; j < numberRecordsPerRow; j++) { mutation.addToRecordMutations(getRecordMutation(numberOfColumns, numberOfFamilies, numberOfWords)); + countRecord++; } - client.mutate(mutation); - count++; + if (batch == 1) { + client.mutate(mutation); + } else { + mutations.add(mutation); + if (mutations.size() >= batch) { + client.mutateBatch(mutations); + mutations.clear(); + } + } + countRow++; } + client.mutateBatch(mutations); + printPerformance(out, countRow, countRecord, start, s, numberRows); + } + + private static void printPerformance(PrintWriter out, int countRow, int countRecord, final long start, long s, int i) { + double totalSeconds = (System.currentTimeMillis() - start) / 1000.0; + double seconds = (System.currentTimeMillis() - s) / 1000.0; + double recordRate = countRecord / seconds; + double rowRate = countRow / seconds; + double avgRowRate = i / totalSeconds; + out.printf("Rows indexed [%d] at Avg Rows [%f/s] Rows [%f/s] Records [%f/s]%n", i, avgRowRate, rowRate, + recordRate); + out.flush(); } private static void loadWords() throws IOException { @@ -119,7 +160,7 @@ public class LoadData { } private static String getWord() { - return makeUpperCaseRandomly(words.get(random.nextInt(words.size())),random); + return makeUpperCaseRandomly(words.get(random.nextInt(words.size())), random); } private static String makeUpperCaseRandomly(String s, Random r) {
