Improvements for HBase and Cassandra clients

.gitignore

@@ -1,5 +1,6 @@
 # ignore compiled byte code
 target
+*.pyc
 
 # ignore output files from testing
 output*
@@ -8,3 +9,7 @@ output*
 .project
 .classpath
 .settings
+
+# and intellij
+.idea/
+*.iml

CHANGELOG

@@ -1,4 +1,18 @@
-- gh-95 Bump MongoDB version to 2.9.0 (allanbank)
+- Upgraded MongoDB client to v.2.10.1 (thumbtack)
+- Added mongodb.readPreference property (thumbtack)
+- Added Aerospike client (thumbtack)
+- Throttling improvement: the throughput of last 100ms is used (thumbtack)
+- Output improvements, exportmeasurementsinterval property (thumbtack)
+- Added warm-up phase before run the workload (thumbtack)
+- Added fieldnameprefix property (thumbtack)
+- Added retries: readretrycount, updateretrycount, insertretrycount, retrydelay
+  properties (thumbtack)
+- Added ignoreinserterrors property (thumbtack)
+- Upgraded Couchbase client to v.1.1.0 and Spymemcached client to v.2.8.9
+  (thumbtack)
+- Added couchbase.replicateTo property (thumbtack)
+- Added Fabric scripts to run YCSB on multiple hosts (thumbtack) 
+
 - gh-67 Use checkstyle (m1ch1)
 - gh-76 Implemented OrientDB client (lvca)
 - gh-88 YCSB client for Amazon DynamoDB (jananin)

IMPROVEMENTS.txt

@@ -0,0 +1,95 @@
+Here is the list of YCSB improments made by Thumbtack Technology.
+
+Upgrade of MongoDB client
+
+We upgraded the MongoDB driver from version 2.8.0 (appeared before 
+Mongo 2.2 release) to version 2.10.1 and allow the readPreference
+(http://docs.mongodb.org/manual/applications/replication/#read-preference) 
+to be set as a configuration.
+Also, now all the write errors are printed to stderr.
+
+New configuration properties
+  mongodb.readPreference = primary|primaryPreferred|secondary|secondaryPreferred
+
+Improvements of Aerospike client
+
+We added ability to display operations result code in more details: 
+no only success or erroneous, but displaying the actual error type.
+
+Throttling improvements
+
+YCSB allows to limit the throughput, but uses the average throughput for 
+the whole time of the experiment. It causes the throughput peaks after 
+downfalls in failover tests. We modified YCSB to keep the desired throughput 
+on the same level, without peaks, by throttling based on the average throughput 
+on the last 100 ms.
+
+Output improvements
+
+Print current statistics to stderr every 2 secs instead of 10 secs
+Print intermediate statistics (identical to final) to stdout in every 
+configured time interval. To avoid stats loss on YCSB hang or crash
+Print final statistics on YCSB process shutdown
+
+New configuration properties
+  exportmeasurementsinterval: interval time for exporting measurements in out 
+      stream in milliseconds (default: 1000)
+
+Warm-up
+
+Allow to do some number of operations before gathering statistics, to warm-up 
+the database. The length of warm-up can be limited by the number of operations 
+or by the time period.
+
+New configuration properties
+  warmupoperationcount: number of operations in warmup phase, 
+      if zero then don't warmup (default: 0)
+  warmupexecutiontime: execution time of warmup phase in milliseconds, 
+      if zero then don't warmup (default: 0)
+
+Field name
+
+By default YCSB names the database record fields as “field” + a number. 
+The new configuration option allows to replace the “field” prefix with 
+something shorter which affects the data storage size for schema-less databases.
+
+New configuration properties
+  fieldnameprefix: string prefix for the field name (default: “field”)
+
+Retries
+
+Added ability to retry failed operations. The max number of retries is limited. 
+The total number of retries is counted. The retries are done within the same 
+operation, so they doesn’t affect the number of operations, but increases 
+the latency of the operation. The retries can be done with delays between them.
+Original YCSB stops on error on insert, now it’s possible to retry the insert.
+
+New configuration properties
+  readretrycount: number of retries if read fails, 
+      if zero then don't retry (default: 0)
+  updateretrycount: number of retries if update fails, 
+      if zero then don't retry (default: 0)
+  insertretrycount: number of retries if insert fails, 
+      if zero then don't retry (default: 0)
+  retrydelay: delay between retries in milliseconds (default: 0)
+
+Inserts with errors
+
+The new configuration option was added to allow errors on inserts.
+Usually YCSB stops when any operation fails on load phase. But now it’s possible 
+to ignore such errors and continue inserting. This can be useful for MongoDB, 
+for example, to rerun the load phase without clearing the data.
+
+New configuration properties
+  ignoreinserterrors: set to true to activate the new feature
+
+Upgrade of Couchbase client
+
+We upgraded the Couchbase driver from version 1.1-dp2 to 1.1.0 and Spymemcached
+driver from version 2.8.4 to 2.8.9 and allow the replicateTo
+(http://www.couchbase.com/autodocs/couchbase-java-client-1.1.0/com/couchbase/client/CouchbaseClient.html#replace(java.lang.String, int, java.lang.String, net.spy.memcached.PersistTo, net.spy.memcached.ReplicateTo)) 
+to be set as a configuration.
+
+New configuration properties
+  couchbase.replicateTo = ZERO|ONE|TWO|THREE
+

README.md

@@ -1,10 +1,23 @@
 Yahoo! Cloud System Benchmark (YCSB)
 ====================================
-[![Build Status](https://travis-ci.org/brianfrankcooper/YCSB.png?branch=master)](https://travis-ci.org/brianfrankcooper/YCSB)
+
+A note on comparing multiple systems
+------------------------------------
+
+NoSQL systems have widely varying defaults for trading off write durability vs performance.  Make sure that you are [comparing apples to apples across all candidates](http://www.datastax.com/dev/blog/how-not-to-benchmark-cassandra-a-case-study).  The most useful common denominator is synchronously durable writes.  The following YCSB clients have been verified to perform synchronously durable writes by default:
+
+- Couchbase
+- HBase
+- MongoDB
+
+Cassandra requires a configuration change in conf/cassandra.yaml.  Uncomment these lines:
+
+    # commitlog_sync: batch
+    # commitlog_sync_batch_window_in_ms: 50
 
 Links
 -----
-http://wiki.github.com/brianfrankcooper/YCSB/  
+http://wiki.github.com/jbellis/YCSB/  
 http://research.yahoo.com/Web_Information_Management/YCSB/  
 ycsb-users@yahoogroups.com  
 
@@ -14,7 +27,7 @@ Getting Started
 1. Download the latest release of YCSB:
 
     ```sh
-    wget https://github.com/downloads/brianfrankcooper/YCSB/ycsb-0.1.4.tar.gz
+    wget https://github.com/downloads/jbellis/YCSB/ycsb-0.1.4.tar.gz
     tar xfvz ycsb-0.1.4
     cd ycsb-0.1.4
     ```
@@ -36,3 +49,6 @@ Getting Started
 
   See https://github.com/brianfrankcooper/YCSB/wiki/Core-Properties for 
   the list of available workload properties.
+
+  Alternatively, see fabric/README for Thumbtack's work on parallelizing
+  YCSB clients using Fabric.

__init__.py

@@ -0,0 +1,3 @@
+__author__ = 'nick'
+# this file is needed to make relative imports possible in Python
+# specifically to access ./conf/* from ./fabfile/*

accumulo/src/main/java/com/yahoo/ycsb/db/AccumuloClient.java

@@ -3,12 +3,12 @@
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Hashtable;
+import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Random;
 import java.util.Set;
 import java.util.TreeSet;
-import java.util.Vector;
 import java.util.concurrent.TimeUnit;
 
 import org.apache.accumulo.core.client.AccumuloException;
@@ -171,8 +171,16 @@ private Scanner getRow(Text row, Set<String> fields)
 	}
 
 	@Override
-	public int read(String table, String key, Set<String> fields,
-			HashMap<String, ByteIterator> result) {
+	public int readOne(String table, String key, String field, Map<String,ByteIterator> result) {
+		return read(table, key, result);
+	}
+
+	@Override
+	public int readAll(String table, String key, Map<String,ByteIterator> result) {
+		return read(table, key, result);
+	}
+
+	public int read(String table, String key, Map<String, ByteIterator> result) {
 
 		try {
 			checkTable(table);
@@ -198,8 +206,26 @@ public int read(String table, String key, Set<String> fields,
 	}
 
 	@Override
+	public int scanOne(String table, String startkey, int recordcount, String field, List<Map<String, ByteIterator>> result) {
+
+		_scanScanner.clearColumns();
+		_scanScanner.setRange(new Range(new Text(startkey), null));
+		_scanScanner.fetchColumn(_colFam, new Text(field));
+
+		return scan(table, startkey, recordcount, result);
+	}
+
+	@Override
+	public int scanAll(String table, String startkey, int recordcount, List<Map<String, ByteIterator>> result) {
+
+		_scanScanner.clearColumns();
+		_scanScanner.setRange(new Range(new Text(startkey), null));
+
+		return scan(table, startkey, recordcount, result);
+	}
+
 	public int scan(String table, String startkey, int recordcount,
-			Set<String> fields, Vector<HashMap<String, ByteIterator>> result) {
+			List<Map<String, ByteIterator>> result) {
 		try {
 			checkTable(table);
 		} catch (TableNotFoundException e) {
@@ -214,19 +240,11 @@ public int scan(String table, String startkey, int recordcount,
 		_scanScanner.setRange(new Range(new Text(startkey), null));
 
 		// Batch size is how many key/values to try to get per call. Here, I'm
-		// guessing that the number of keys in a row is equal to the number of fields 
+		// guessing that the number of keys in a row is equal to the number of fields
 		// we're interested in.
-		// We try to fetch one more so as to tell when we've run out of fields.
+		// We try to fetch one or more using either ScanOne or scanAll respectively so
+		// as to tell when we've run out of fields.
 
-		if (fields != null) {
-			// And add each of them as fields we want.
-			for(String field:fields)
-			{
-				_scanScanner.fetchColumn(_colFam, new Text(field));
-			}
-		} else {
-			// If no fields are provided, we assume one column/row.
-		}
 
 		String rowKey = "";
 		HashMap<String, ByteIterator> currentHM = null;
@@ -240,15 +258,8 @@ public int scan(String table, String startkey, int recordcount,
 					break;
 				}
 				rowKey = entry.getKey().getRow().toString();
-				if (fields != null) {
-					// Initial Capacity for all keys.
-					currentHM = new HashMap<String, ByteIterator>(fields.size()); 
-				}
-				else
-				{
-					// An empty result map.
-					currentHM = new HashMap<String, ByteIterator>();
-				}
+                currentHM = new HashMap<String, ByteIterator>();
+
 				result.add(currentHM);
 			}
 			// Now add the key to the hashmap.
@@ -262,13 +273,17 @@ public int scan(String table, String startkey, int recordcount,
 	}
 
 	@Override
-	public int update(String table, String key, HashMap<String, ByteIterator> values) {
-		try {
-			checkTable(table);
-		} catch (TableNotFoundException e) {
-			System.err.println("Error trying to connect to Accumulo table." + e);
-			return ServerError;
-		}
+	public int updateOne(String table, String key, String field, ByteIterator value) {
+
+		Mutation mutInsert = new Mutation(new Text(key));
+		mutInsert.put(_colFam, new Text(field), System.currentTimeMillis(),
+				new Value(value.toArray()));
+
+		return update(table, key, mutInsert);
+	}
+
+	@Override
+	public int updateAll(String table, String key, Map<String,ByteIterator> values) {
 
 		Mutation mutInsert = new Mutation(new Text(key));
 		for (Map.Entry<String, ByteIterator> entry : values.entrySet()) {
@@ -277,6 +292,17 @@ public int update(String table, String key, HashMap<String, ByteIterator> values
 					new Value(entry.getValue().toArray()));
 		}
 
+		return update(table, key, mutInsert);
+	}
+
+	public int update(String table, String key, /*Map<String, ByteIterator> values*/Mutation mutInsert) {
+		try {
+			checkTable(table);
+		} catch (TableNotFoundException e) {
+			System.err.println("Error trying to connect to Accumulo table." + e);
+			return ServerError;
+		}
+
 		try {
 			_bw.addMutation(mutInsert);
 			// Distributed YCSB co-ordination: YCSB on a client produces the key to 
@@ -296,8 +322,8 @@ public int update(String table, String key, HashMap<String, ByteIterator> values
 	}
 
 	@Override
-	public int insert(String table, String key, HashMap<String, ByteIterator> values) {
-		return update(table, key, values);
+	public int insert(String table, String key, Map<String, ByteIterator> values) {
+		return updateAll(table, key, values);
 	}
 
 	@Override
@@ -381,7 +407,7 @@ private void keyNotification(String key) {
 						fields.add("field"+j);
 					HashMap<String,ByteIterator> result = new HashMap<String,ByteIterator>();
 
-					int retval = read(table, strKey, fields, result);
+					int retval = read(table, strKey, result);
 					//If the results are empty, the key is enqueued in Zookeeper
 					//and tried again, until the results are found.
 					if (result.size() == 0) {