Spring/spring-data-redis
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Implement Map/Reduce for AsyncRiakTemplate, Groovy DSL, nesting operations into a node that serves as a default bucket.

Browse Source
This commit is contained in:
J. Brisbin
2010-12-22 16:40:19 -06:00
parent e661d22104
commit 8e75fbfaf0
12 changed files with 716 additions and 212 deletions
Download Patch File Download Diff File Expand all files Collapse all files

51
spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/core/AsyncRiakTemplate.java
View File

@@ -22,8 +22,11 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.dao.DataAccessResourceFailureException;
import org.springframework.data.keyvalue.riak.DataStoreOperationException;
import org.springframework.data.keyvalue.riak.mapreduce.AsyncMapReduceOperations;
import org.springframework.data.keyvalue.riak.mapreduce.MapReduceJob;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.http.client.ClientHttpRequestFactory;
import org.springframework.util.Assert;
@@ -43,7 +46,7 @@ import java.util.concurrent.Future;
/**
* @author J. Brisbin <jon@jbrisbin.com>
*/
public class AsyncRiakTemplate extends AbstractRiakTemplate implements AsyncBucketKeyValueStoreOperations {
public class AsyncRiakTemplate extends AbstractRiakTemplate implements AsyncBucketKeyValueStoreOperations, AsyncMapReduceOperations {
protected final Logger log = LoggerFactory.getLogger(getClass());
@@ -301,6 +304,17 @@ public class AsyncRiakTemplate extends AbstractRiakTemplate implements AsyncBuck
return setWithMetaData(bucket, key, value, metaData, null, callback);
}
/* ---------------- Map/Reduce ---------------- */
@SuppressWarnings({"unchecked"})
public Future<?> execute(MapReduceJob job, AsyncKeyValueStoreOperation<List<?>> callback) {
HttpHeaders headers = defaultHeaders(null);
headers.setContentType(MediaType.APPLICATION_JSON);
HttpEntity<String> json = new HttpEntity<String>(job.toJson(), headers);
return workerPool.submit(new AsyncMapReduce(json, callback));
}
/* ---------------- Runnable helpers ---------------- */
protected class AsyncPut<V> implements Runnable {
private String bucket;
@@ -384,6 +398,41 @@ public class AsyncRiakTemplate extends AbstractRiakTemplate implements AsyncBuck
}
protected class AsyncMapReduce implements Runnable {
private HttpEntity<String> entity = null;
private AsyncKeyValueStoreOperation<List<?>> callback = null;
public AsyncMapReduce(HttpEntity<String> entity, AsyncKeyValueStoreOperation<List<?>> callback) {
this.entity = entity;
this.callback = callback;
}
@SuppressWarnings({"unchecked"})
public void run() {
try {
HttpEntity<List> result = getRestTemplate().postForEntity(mapReduceUri,
entity,
List.class);
if (log.isDebugEnabled()) {
log.debug(String.format("M/R: json=%s", entity.getBody()));
}
if (null != callback) {
RiakMetaData meta = extractMetaData(result.getHeaders());
callback.completed(meta, result.getBody());
}
} catch (Throwable t) {
DataStoreOperationException dsoe = new DataStoreOperationException(t.getMessage(), t);
if (null != callback) {
callback.failed(dsoe);
} else {
defaultErrorHandler.failed(dsoe);
}
}
}
}
protected class AsyncGet<T> implements Runnable {
private String bucket;

298
spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/groovy/RiakBuilder.java
View File

@@ -26,7 +26,11 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.keyvalue.riak.DataStoreOperationException;
import org.springframework.data.keyvalue.riak.core.AsyncRiakTemplate;
import org.springframework.data.keyvalue.riak.core.RiakQosParameters;
import org.springframework.data.keyvalue.riak.core.SimpleBucketKeyPair;
import org.springframework.data.keyvalue.riak.mapreduce.*;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -37,10 +41,14 @@ import java.util.concurrent.Executors;
public class RiakBuilder extends BuilderSupport {
protected final Logger log = LoggerFactory.getLogger(getClass());
@Autowired
@Autowired(required = false)
protected AsyncRiakTemplate riak;
@Autowired
@Autowired(required = false)
protected ExecutorService workerPool = Executors.newCachedThreadPool();
protected String defaultBucketName;
public RiakBuilder() {
}
public RiakBuilder(AsyncRiakTemplate riak) {
this.riak = riak;
@@ -61,6 +69,14 @@ public class RiakBuilder extends BuilderSupport {
this.riak = riak;
}
public AsyncRiakTemplate getAsyncTemplate() {
return riak;
}
public void setAsyncTemplate(AsyncRiakTemplate riak) {
this.riak = riak;
}
public ExecutorService getWorkerPool() {
return workerPool;
}
@@ -74,15 +90,65 @@ public class RiakBuilder extends BuilderSupport {
log.debug("setParent/2 " + parent + " " + child);
}
@SuppressWarnings({"unchecked"})
@Override
protected Object createNode(Object name) {
log.debug("createNode/1 " + name);
return this;
if ("call".equals(name)) {
// IGNORED
} else if ("foreach".equals(name)) {
RiakOperation<Object> op = new RiakOperation<Object>(riak, RiakOperation.Type.FOREACH);
op.setBucket(defaultBucketName);
return op;
} else if ("mapreduce".equals(name)) {
return createMapReduceJob();
} else if ("query".equals(name)) {
QueryPhase p = new QueryPhase();
p.job = ((RiakMapReduceOperation) getCurrent()).getJob();
return getCurrent();
} else if ("map".equals(name) || "reduce".equals(name)) {
QueryPhase p = new QueryPhase();
p.job = ((RiakMapReduceOperation) getCurrent()).getJob();
p.phase = name.toString();
return p;
} else {
defaultBucketName = name.toString();
}
return null;
}
@SuppressWarnings({"unchecked"})
@Override
protected Object createNode(Object name, Object value) {
log.debug("createNode/2 " + name + " " + value);
if ("inputs".equals(name)) {
AsyncRiakMapReduceJob job = ((RiakMapReduceOperation) getCurrent()).getJob();
if (null != value && value instanceof String) {
List<String> keys = new ArrayList<String>();
keys.add(value.toString());
job.addInputs(keys);
} else if (value instanceof List) {
job.addInputs((List) value);
}
return job;
} else if ("language".equals(name)) {
QueryPhase p = (QueryPhase) getCurrent();
p.language = value.toString();
return p;
} else if ("source".equals(name)) {
QueryPhase p = (QueryPhase) getCurrent();
p.source = value.toString();
return p;
} else if ("keep".equals(name)) {
QueryPhase p = (QueryPhase) getCurrent();
p.keep = (value instanceof Boolean ? (Boolean) value : new Boolean(value.toString()));
return p;
} else if ("arg".equals(name)) {
QueryPhase p = (QueryPhase) getCurrent();
p.arg = value;
return p;
}
return null; //To change body of implemented methods use File | Settings | File Templates.
}
@@ -90,59 +156,96 @@ public class RiakBuilder extends BuilderSupport {
@Override
protected Object createNode(Object name, Map attributes) {
log.debug("createNode/2 (Map) " + name + " " + attributes);
RiakOperation.Type type = RiakOperation.Type.valueOf(name.toString().toUpperCase());
if (null != type) {
RiakOperation op = new RiakOperation<Object>(riak, type);
Object o = attributes.get("bucket");
op.setBucket((null != o ? o.toString() : null));
o = attributes.get("key");
op.setKey((null != o ? o.toString() : null));
o = attributes.get("value");
op.setValue(o);
o = attributes.get("type");
if (null != o) {
if (o instanceof Class) {
op.setRequiredType((Class<?>) o);
} else if (o instanceof String) {
try {
op.setRequiredType(Class.forName((String) o));
} catch (ClassNotFoundException e) {
throw new DataStoreOperationException(e.getMessage(), e);
}
} else {
op.setRequiredType(o.getClass());
}
}
o = attributes.get("qos");
if (null != o) {
RiakQosParameters qos = new RiakQosParameters();
Map<String, Object> qosParams = (Map<String, Object>) o;
if (qosParams.containsKey("dw")) {
qos.setDurableWriteThreshold(qosParams.get("dw"));
}
if (qosParams.containsKey("w")) {
qos.setWriteThreshold(qosParams.get("w"));
}
if (qosParams.containsKey("r")) {
qos.setReadThreshold(qosParams.get("r"));
}
op.setQosParameters(qos);
}
o = attributes.get("wait");
if ("mapreduce".equals(name)) {
RiakMapReduceOperation oper = createMapReduceJob();
// Set timeout
Object o = attributes.get("wait");
if (null != o) {
if (o instanceof Long) {
op.setTimeout((Long) o);
oper.setTimeout((Long) o);
} else if (o instanceof String) {
op.setTimeout(new Long(o.toString()));
oper.setTimeout(new Long(o.toString()));
} else if (o instanceof Integer) {
op.setTimeout(new Long((Integer) o));
oper.setTimeout(new Long((Integer) o));
} else {
throw new IllegalArgumentException(
"Timeout should be an Integer, a Long, or a String denoting milliseconds");
}
}
return op;
return oper;
} else if ("map".equals(name) || "reduce".equals(name)) {
QueryPhase p = new QueryPhase();
p.job = ((RiakMapReduceOperation) getCurrent()).getJob();
p.phase = name.toString();
// Set arg
p.arg = attributes.get("arg");
return p;
} else {
RiakOperation.Type type = RiakOperation.Type.valueOf(name.toString().toUpperCase());
if (null != type) {
RiakOperation op = new RiakOperation<Object>(riak, type);
// Set a bucket name
Object o = attributes.get("bucket");
if (null == o && null != defaultBucketName) {
op.setBucket(defaultBucketName);
} else {
op.setBucket((null != o ? o.toString() : null));
}
// Set the object's key
o = attributes.get("key");
op.setKey((null != o ? o.toString() : null));
// Set the value
o = attributes.get("value");
op.setValue(o);
// Set the type of object (for getAsType)
o = attributes.get("type");
if (null != o) {
if (o instanceof Class) {
op.setRequiredType((Class<?>) o);
} else if (o instanceof String) {
try {
op.setRequiredType(Class.forName((String) o));
} catch (ClassNotFoundException e) {
throw new DataStoreOperationException(e.getMessage(), e);
}
} else {
op.setRequiredType(o.getClass());
}
}
// Set QOS parameters
o = attributes.get("qos");
if (null != o) {
RiakQosParameters qos = new RiakQosParameters();
Map<String, Object> qosParams = (Map<String, Object>) o;
if (qosParams.containsKey("dw")) {
qos.setDurableWriteThreshold(qosParams.get("dw"));
}
if (qosParams.containsKey("w")) {
qos.setWriteThreshold(qosParams.get("w"));
}
if (qosParams.containsKey("r")) {
qos.setReadThreshold(qosParams.get("r"));
}
op.setQosParameters(qos);
}
// Set timeout
o = attributes.get("wait");
if (null != o) {
if (o instanceof Long) {
op.setTimeout((Long) o);
} else if (o instanceof String) {
op.setTimeout(new Long(o.toString()));
} else if (o instanceof Integer) {
op.setTimeout(new Long((Integer) o));
} else {
throw new IllegalArgumentException(
"Timeout should be an Integer, a Long, or a String denoting milliseconds");
}
}
return op;
}
}
return null;
}
@@ -163,38 +266,45 @@ public class RiakBuilder extends BuilderSupport {
@Override
public Object invokeMethod(String methodName, Object arg) {
if (log.isDebugEnabled()) {
log.debug("invokeMethod: " + methodName + " " + arg);
log.debug("invokeMethod/2: " + methodName + " " + arg);
}
Object[] args = (Object[]) arg;
Map<String, Object> params;
Closure handler = null;
RiakOperation<Object> op;
if ("completed".equals(methodName) || "failed".equals(methodName)) {
op = (RiakOperation<Object>) getCurrent();
Closure guard = null;
for (Object o : args) {
if (o instanceof Map) {
params = (Map<String, Object>) o;
if (params.containsKey("when")) {
guard = (Closure) params.get("when");
if (getCurrent() instanceof RiakOperation) {
RiakOperation<Object> op = (RiakOperation<Object>) getCurrent();
Object[] args = (Object[]) arg;
Map<String, Object> params;
Closure handler = null;
Closure guard = null;
for (Object o : args) {
if (o instanceof Map) {
params = (Map<String, Object>) o;
if (params.containsKey("when")) {
guard = (Closure) params.get("when");
}
} else if (o instanceof Closure) {
handler = (Closure) o;
}
} else if (o instanceof Closure) {
handler = (Closure) o;
}
op.addHandler(methodName, handler, guard);
return op;
} else if (getCurrent() instanceof RiakMapReduceOperation) {
RiakMapReduceOperation oper = (RiakMapReduceOperation) getCurrent();
Object[] args = (Object[]) arg;
if ("completed".equals(methodName)) {
oper.setCompleted((Closure) args[0]);
} else if ("failed".equals(methodName)) {
oper.setFailed((Closure) args[0]);
}
return oper;
}
op.addHandler(methodName, handler, guard);
return op;
}
return super.invokeMethod(methodName, arg);
}
@SuppressWarnings({"unchecked"})
@Override
protected void nodeCompleted(Object parent, Object node) {
log.debug("nodeCompleted: " + parent + " " + node);
log.debug("nodeCompleted: parent=" + parent + ", node=" + node);
if (node instanceof RiakOperation) {
RiakOperation<Object> op = (RiakOperation<Object>) node;
try {
@@ -202,6 +312,26 @@ public class RiakBuilder extends BuilderSupport {
} catch (Exception e) {
log.error(e.getMessage(), e);
}
} else if (parent instanceof RiakMapReduceOperation && node instanceof QueryPhase) {
QueryPhase p = (QueryPhase) node;
MapReduceOperation oper = null;
if ("javascript".equals(p.language)) {
if (null != p.source) {
oper = new JavascriptMapReduceOperation(p.source);
} else if (null != p.bucket && null != p.key) {
oper = new JavascriptMapReduceOperation(new SimpleBucketKeyPair(p.bucket, p.key));
}
} else {
oper = new ErlangMapReduceOperation(p.module, p.func);
}
if (null != oper) {
RiakMapReducePhase phase = new RiakMapReducePhase(p.phase, p.language, oper);
if (null != p.keep) {
phase.setKeepResults(p.keep);
}
phase.setArg(p.arg);
p.job.addPhase(phase);
}
} else {
super.nodeCompleted(parent, node);
}
@@ -211,7 +341,43 @@ public class RiakBuilder extends BuilderSupport {
@Override
protected Object postNodeCompletion(Object parent, Object node) {
log.debug("postNodeCompletion: " + parent + " " + node);
if (null == parent && node instanceof RiakMapReduceOperation) {
RiakMapReduceOperation oper = (RiakMapReduceOperation) node;
try {
return oper.call();
} catch (Exception e) {
log.error(e.getMessage(), e);
}
} else if (null == parent && node == parent) {
defaultBucketName = null;
}
return super.postNodeCompletion(parent, node);
}
protected RiakMapReduceOperation createMapReduceJob() {
AsyncRiakMapReduceJob job = new AsyncRiakMapReduceJob(riak);
if (null != defaultBucketName) {
List<String> keys = new ArrayList<String>();
keys.add(defaultBucketName);
job.addInputs(keys);
}
return new RiakMapReduceOperation(riak, job);
}
private class QueryPhase {
AsyncRiakMapReduceJob job;
String phase;
String language = "javascript";
String source = null;
String bucket = null;
String key = null;
String module = null;
String func = null;
Boolean keep = null;
Object arg = null;
}
}

110
spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/groovy/RiakMapReduceOperation.java Normal file
View File

@@ -0,0 +1,110 @@
/*
* Copyright (c) 2010 by J. Brisbin <jon@jbrisbin.com>
* Portions (c) 2010 by NPC International, Inc. or the
* original author(s).
*
* Licensed 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.
*/
package org.springframework.data.keyvalue.riak.groovy;
import groovy.lang.Closure;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.keyvalue.riak.core.AsyncKeyValueStoreOperation;
import org.springframework.data.keyvalue.riak.core.AsyncRiakTemplate;
import org.springframework.data.keyvalue.riak.core.KeyValueStoreMetaData;
import org.springframework.data.keyvalue.riak.mapreduce.AsyncRiakMapReduceJob;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
/**
* @author J. Brisbin <jon@jbrisbin.com>
*/
public class RiakMapReduceOperation implements Callable {
protected final Logger log = LoggerFactory.getLogger(getClass());
protected AsyncRiakTemplate riak;
protected AsyncRiakMapReduceJob job;
protected Long timeout = -1L;
protected Closure completed = null;
protected Closure failed = null;
public RiakMapReduceOperation(AsyncRiakTemplate riak, AsyncRiakMapReduceJob job) {
this.riak = riak;
this.job = job;
}
public AsyncRiakMapReduceJob getJob() {
return job;
}
public void setJob(AsyncRiakMapReduceJob job) {
this.job = job;
}
public Long getTimeout() {
return timeout;
}
public void setTimeout(Long timeout) {
this.timeout = timeout;
}
public Closure getCompleted() {
return completed;
}
public void setCompleted(Closure completed) {
this.completed = completed;
}
public Closure getFailed() {
return failed;
}
public void setFailed(Closure failed) {
this.failed = failed;
}
public Object call() throws Exception {
Future<?> f = riak.execute(job, new AsyncKeyValueStoreOperation<List<?>>() {
public void completed(KeyValueStoreMetaData meta, List<?> result) {
Object arg = new Object[]{result, meta};
if (null != completed) {
completed.call(arg);
}
}
public void failed(Throwable error) {
if (null != failed) {
failed.call(error);
} else {
throw new RuntimeException(error);
}
}
});
if (timeout == 0) {
return f;
} else if (timeout < 0) {
return f.get();
} else {
return f.get(timeout, TimeUnit.MILLISECONDS);
}
}
}

4
spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/groovy/RiakOperation.java
View File

@@ -39,7 +39,7 @@ import java.util.concurrent.*;
public class RiakOperation<T> implements Callable {
static enum Type {
SET, SETASBYTES, PUT, GET, GETASBYTES, GETASTYPE, CONTAINSKEY, DELETE, EACH
SET, SETASBYTES, PUT, GET, GETASBYTES, GETASTYPE, CONTAINSKEY, DELETE, FOREACH
}
static String COMPLETED = "completed";
@@ -162,7 +162,7 @@ public class RiakOperation<T> implements Callable {
case DELETE:
f = riak.delete(bucket, key, callbackInvoker);
break;
case EACH:
case FOREACH:
f = riak.getBucketSchema(bucket,
null,
new AsyncKeyValueStoreOperation<Map<String, Object>>() {

142
spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/mapreduce/AbstractRiakMapReduceJob.java Normal file
View File

@@ -0,0 +1,142 @@
/*
* Copyright (c) 2010 by J. Brisbin <jon@jbrisbin.com>
* Portions (c) 2010 by NPC International, Inc. or the
* original author(s).
*
* Licensed 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.
*/
package org.springframework.data.keyvalue.riak.mapreduce;
import org.codehaus.jackson.JsonFactory;
import org.codehaus.jackson.JsonGenerator;
import org.codehaus.jackson.map.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.keyvalue.riak.core.BucketKeyPair;
import java.io.IOException;
import java.io.StringWriter;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
/**
* An implementation of {@link MapReduceJob} for the Riak data store.
*
* @author J. Brisbin <jon@jbrisbin.com>
*/
@SuppressWarnings({"unchecked"})
public abstract class AbstractRiakMapReduceJob implements MapReduceJob {
protected final Logger log = LoggerFactory.getLogger(getClass());
protected List<Object> inputs = new LinkedList<Object>();
protected List<MapReducePhase> phases = new ArrayList<MapReducePhase>();
public List getInputs() {
return this.inputs;
}
public MapReduceJob addInputs(List keys) {
inputs.addAll(keys);
return this;
}
public MapReduceJob addPhase(MapReducePhase phase) {
phases.add(phase);
return this;
}
public List<MapReducePhase> getPhases() {
return this.phases;
}
public String toJson() {
StringWriter out = new StringWriter();
try {
JsonGenerator json = new JsonFactory().createJsonGenerator(out);
json.setCodec(new ObjectMapper());
json.writeStartObject();
// Inputs
json.writeFieldName("inputs");
if (1 == inputs.size() && !(inputs.get(0) instanceof List)) {
json.writeString(inputs.get(0).toString());
} else if (inputs.size() > 0) {
json.writeStartArray();
for (Object obj : inputs) {
List pair = (List) obj;
json.writeStartArray();
json.writeString(pair.get(0).toString());
json.writeString(pair.get(1).toString());
json.writeEndArray();
}
json.writeEndArray();
}
// Query
json.writeFieldName("query");
json.writeStartArray();
for (MapReducePhase phase : phases) {
json.writeStartObject();
switch (phase.getPhase()) {
case MAP:
json.writeFieldName("map");
break;
case REDUCE:
json.writeFieldName("reduce");
break;
}
json.writeStartObject();
json.writeStringField("language", phase.getLanguage());
Object repr = phase.getOperation().getRepresentation();
if (repr instanceof String) {
// Using source
json.writeStringField("source",
String.format("%s", phase.getOperation().getRepresentation()));
} else if (repr instanceof BucketKeyPair) {
BucketKeyPair pair = (BucketKeyPair) repr;
json.writeStringField("bucket",
String.format("%s", pair.getBucket()));
json.writeStringField("key", String.format("%s", pair.getKey()));
} else if (repr instanceof Map) {
for (Map.Entry<Object, Object> entry : ((Map<Object, Object>) repr).entrySet()) {
json.writeStringField(entry.getKey().toString(),
entry.getValue().toString());
}
}
if (phase.getKeepResults()) {
json.writeBooleanField("keep", true);
}
// Arg
if (null != phase.getArg()) {
json.writeObjectField("arg", phase.getArg());
}
json.writeEndObject();
json.writeEndObject();
}
json.writeEndArray();
json.writeEndObject();
json.flush();
} catch (IOException e) {
log.error(e.getMessage(), e);
}
return out.toString();
}
}

41
spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/mapreduce/AsyncMapReduceOperations.java Normal file
View File

@@ -0,0 +1,41 @@
/*
* Copyright (c) 2010 by J. Brisbin <jon@jbrisbin.com>
* Portions (c) 2010 by NPC International, Inc. or the
* original author(s).
*
* Licensed 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.
*/
package org.springframework.data.keyvalue.riak.mapreduce;
import org.springframework.data.keyvalue.riak.core.AsyncKeyValueStoreOperation;
import java.util.List;
import java.util.concurrent.Future;
/**
* Generic interface to Map/Reduce in data stores that support it.
*
* @author J. Brisbin <jon@jbrisbin.com>
*/
public interface AsyncMapReduceOperations {
/**
* Execute a {@link MapReduceJob} synchronously.
*
* @param job
* @return
*/
Future<?> execute(MapReduceJob job, AsyncKeyValueStoreOperation<List<?>> callback);
}

48
spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/mapreduce/AsyncRiakMapReduceJob.java Normal file
View File

@@ -0,0 +1,48 @@
/*
* Copyright (c) 2010 by J. Brisbin <jon@jbrisbin.com>
* Portions (c) 2010 by NPC International, Inc. or the
* original author(s).
*
* Licensed 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.
*/
package org.springframework.data.keyvalue.riak.mapreduce;
import org.springframework.data.keyvalue.riak.core.AsyncRiakTemplate;
/**
* An implementation of {@link MapReduceJob} for the Riak data store.
*
* @author J. Brisbin <jon@jbrisbin.com>
*/
@SuppressWarnings({"unchecked"})
public class AsyncRiakMapReduceJob extends AbstractRiakMapReduceJob {
protected AsyncRiakTemplate riakTemplate;
public AsyncRiakMapReduceJob(AsyncRiakTemplate riakTemplate) {
this.riakTemplate = riakTemplate;
}
public AsyncRiakTemplate getAsyncRiakTemplate() {
return riakTemplate;
}
public void setAsyncRiakTemplate(AsyncRiakTemplate riakTemplate) {
this.riakTemplate = riakTemplate;
}
public Object call() throws Exception {
return riakTemplate.execute(this, null);
}
}

15
spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/mapreduce/MapReduceJob.java
View File

@@ -60,21 +60,6 @@ public interface MapReduceJob<T> extends Callable {
*/
List<MapReducePhase> getPhases();
/**
* Set the static argument for this job.
*
* @param arg
*/
void setArg(T arg);
/**
* Get the static argument for this job.
*
* @param <T>
* @return
*/
<T> T getArg();
/**
* Convert this job into the appropriate JSON to send to the server.
*

13
spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/mapreduce/MapReducePhase.java
View File

@@ -52,4 +52,17 @@ public interface MapReducePhase {
*/
MapReduceOperation getOperation();
/**
* Set the static argument for this job.
*
* @param arg
*/
void setArg(Object arg);
/**
* Get the static argument for this phase.
*
* @return
*/
Object getArg();
}

120
spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/mapreduce/RiakMapReduceJob.java
View File

@@ -18,20 +18,8 @@
package org.springframework.data.keyvalue.riak.mapreduce;
import org.codehaus.jackson.JsonFactory;
import org.codehaus.jackson.JsonGenerator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.keyvalue.riak.core.BucketKeyPair;
import org.springframework.data.keyvalue.riak.core.RiakTemplate;
import java.io.IOException;
import java.io.StringWriter;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
/**
* An implementation of {@link org.springframework.data.keyvalue.riak.mapreduce.MapReduceJob}
* for the Riak data store.
@@ -39,12 +27,8 @@ import java.util.Map;
* @author J. Brisbin <jon@jbrisbin.com>
*/
@SuppressWarnings({"unchecked"})
public class RiakMapReduceJob implements MapReduceJob {
public class RiakMapReduceJob extends AbstractRiakMapReduceJob {
protected final Logger log = LoggerFactory.getLogger(getClass());
protected List<Object> inputs = new LinkedList<Object>();
protected List<MapReducePhase> phases = new ArrayList<MapReducePhase>();
protected Object arg = null;
protected RiakTemplate riakTemplate;
public RiakMapReduceJob(RiakTemplate riakTemplate) {
@@ -59,108 +43,6 @@ public class RiakMapReduceJob implements MapReduceJob {
this.riakTemplate = riakTemplate;
}
public List getInputs() {
return this.inputs;
}
public MapReduceJob addInputs(List keys) {
inputs.addAll(keys);
return this;
}
public MapReduceJob addPhase(MapReducePhase phase) {
phases.add(phase);
return this;
}
public List<MapReducePhase> getPhases() {
return this.phases;
}
public void setArg(Object arg) {
this.arg = arg;
}
public Object getArg() {
return this.arg;
}
public String toJson() {
StringWriter out = new StringWriter();
try {
JsonGenerator json = new JsonFactory().createJsonGenerator(out);
json.writeStartObject();
// Inputs
json.writeFieldName("inputs");
if (1 == inputs.size() && !(inputs.get(0) instanceof List)) {
json.writeString(inputs.get(0).toString());
} else if (inputs.size() > 0) {
json.writeStartArray();
for (Object obj : inputs) {
List pair = (List) obj;
json.writeStartArray();
json.writeString(pair.get(0).toString());
json.writeString(pair.get(1).toString());
json.writeEndArray();
}
json.writeEndArray();
}
// Query
json.writeFieldName("query");
json.writeStartArray();
for (MapReducePhase phase : phases) {
json.writeStartObject();
switch (phase.getPhase()) {
case MAP:
json.writeFieldName("map");
break;
case REDUCE:
json.writeFieldName("reduce");
break;
}
json.writeStartObject();
json.writeStringField("language", phase.getLanguage());
Object repr = phase.getOperation().getRepresentation();
if (repr instanceof String) {
// Using source
json.writeStringField("source",
String.format("%s", phase.getOperation().getRepresentation()));
} else if (repr instanceof BucketKeyPair) {
BucketKeyPair pair = (BucketKeyPair) repr;
json.writeStringField("bucket",
String.format("%s", pair.getBucket()));
json.writeStringField("key", String.format("%s", pair.getKey()));
} else if (repr instanceof Map) {
for (Map.Entry<Object, Object> entry : ((Map<Object, Object>) repr).entrySet()) {
json.writeStringField(entry.getKey().toString(),
entry.getValue().toString());
}
}
if (phase.getKeepResults()) {
json.writeBooleanField("keep", true);
}
json.writeEndObject();
json.writeEndObject();
}
json.writeEndArray();
// Arg
if (null != arg) {
json.writeObjectField("arg", arg);
}
json.writeEndObject();
json.flush();
} catch (IOException e) {
log.error(e.getMessage(), e);
}
return out.toString();
}
public Object call() throws Exception {
return riakTemplate.execute(this);
}

9
spring-data-riak/src/main/java/org/springframework/data/keyvalue/riak/mapreduce/RiakMapReducePhase.java
View File

@@ -30,6 +30,7 @@ public class RiakMapReducePhase implements MapReducePhase {
protected String language;
protected MapReduceOperation operation;
protected boolean keepResults = false;
protected Object arg;
public RiakMapReducePhase(String phase, String language, MapReduceOperation oper) {
this.phase = Phase.valueOf(phase.toUpperCase());
@@ -67,4 +68,12 @@ public class RiakMapReducePhase implements MapReducePhase {
this.operation = oper;
}
public Object getArg() {
return arg;
}
public void setArg(Object arg) {
this.arg = arg;
}
}

77
spring-data-riak/src/test/groovy/org/springframework/data/keyvalue/riak/core/RiakBuilderSpec.groovy
View File

@@ -147,14 +147,14 @@ class RiakBuilderSpec extends Specification {
}
def "Test builder each"() {
def "Test builder foreach"() {
given:
def riak = new RiakBuilder(riakTemplate)
def idCnt = 0
when:
riak.each(bucket: "test") {
riak.foreach(bucket: "test") {
completed { idCnt++ }
failed { it.printStackTrace() }
}
@@ -176,7 +176,7 @@ class RiakBuilderSpec extends Specification {
put(bucket: "test", value: [test: "value 2"])
put(bucket: "test", value: [test: "value 3"])
each(bucket: "test") {
foreach(bucket: "test") {
completed { v, meta -> ids << meta.key }
failed { it.printStackTrace() }
}
@@ -188,6 +188,61 @@ class RiakBuilderSpec extends Specification {
}
def "Test builder bucket as node"() {
given:
def riak = new RiakBuilder(riakTemplate)
def ids = []
when:
riak {
"test" {
put(value: [test: "value 1"])
put(value: [test: "value 2"])
put(value: [test: "value 3"])
foreach {
completed { v, meta -> ids << meta.key }
failed { it.printStackTrace() }
}
}
}
then:
null != ids
3 <= ids.size()
}
def "Test builder Map/Reduce"() {
given:
def riak = new RiakBuilder(riakTemplate)
def result = []
when:
riak {
mapreduce {
inputs "test"
query {
map(arg: [test: "arg", alist: [1, 2, 3, 4]]) {
source "function(v, keyInfo, arg){ ejsLog('/tmp/mapred.log', JSON.stringify(v)); ejsLog('/tmp/mapred.log', JSON.stringify(keyInfo)); ejsLog('/tmp/mapred.log', JSON.stringify(arg)); return [1]; }"
}
reduce {
source "function(v){ ejsLog('/tmp/mapred.log', JSON.stringify(arguments)); return Riak.reduceSum(v); }"
}
}
completed { result = it }
failed { it.printStackTrace() }
}
}
then:
null != result
1 <= result.size()
}
def "Test builder delete"() {
given:
@@ -195,14 +250,18 @@ class RiakBuilderSpec extends Specification {
def deleted = false
when:
riak.each(bucket: "test") {
completed { v, meta ->
delete(bucket: meta.bucket, key: meta.key) {
completed { deleted = true }
failed { deleted = false }
riak {
"test" {
foreach {
completed { v, meta ->
delete(bucket: meta.bucket, key: meta.key) {
completed { deleted = true }
failed { deleted = false }
}
}
failed { it.printStackTrace() }
}
}
failed { it.printStackTrace() }
}
then: