|
|
|
|
@@ -1,547 +0,0 @@
|
|
|
|
|
<?xml version="1.0" encoding="UTF-8"?>
|
|
|
|
|
<chapter xml:id="riak" xmlns="http://docbook.org/ns/docbook" version="5.0">
|
|
|
|
|
<title>Riak Support</title>
|
|
|
|
|
|
|
|
|
|
<para><ulink url="https://wiki.basho.com/display/RIAK/Riak">Riak</ulink> is a Key/Value datastore that supports <ulink url="https://wiki.basho.com/pages/viewpage.action?pageId=1245320">Internet-scale data replication</ulink> for high performance and high availability. Spring Data Key/Value (SDKV) provides access to the Riak datastore over the <ulink url="https://wiki.basho.com/display/RIAK/REST+API">HTTP REST API</ulink> using a built-in driver based on Spring 3.0's <ulink url="http://static.springsource.org/spring/docs/3.0.x/spring-framework-reference/html/remoting.html#rest-resttemplate"><classname>RestTemplate</classname></ulink>. In addition to making Key/Value datastore access easier from Java, the <classname>RiakTemplate</classname> has been designed, from the ground up, to be used from alternative JVM languages like <ulink url="http://groovy.codehaus.org/">Groovy</ulink> or <ulink url="http://jruby.org/">JRuby</ulink>.</para>
|
|
|
|
|
|
|
|
|
|
<para>Since the SDKV support for Riak uses the stateless REST API, there are no connection factories to manage or other stateful objects to keep tabs on. The helper you'll spend the most time working with is likely the thread-safe <classname>RiakTemplate</classname> or <classname>RiakKeyValueTemplate</classname>. Your choice of which to use will depend on how you want to manage buckets and keys. SDKV supports two ways to interact with Riak. If you want to use the convention you're likely already familiar with, namely of storing an entry with a given key in a "bucket" by passing the bucket and key name separately, you'll want to use the <classname>RiakTemplate</classname>. If you want to use a single object to represent your bucket and key pair, you can use the <classname>RiakKeyValueTemplate</classname>. It supports a key object that is encoded using one of several different methods:
|
|
|
|
|
<itemizedlist>
|
|
|
|
|
<listitem><emphasis>Using a <classname>String</classname></emphasis> - You can concatenate two strings, separated by a colon: "mybucket:mykey".</listitem>
|
|
|
|
|
<listitem><emphasis>Using a <interfacename>BucketKeyPair</interfacename></emphasis> - You can pass an instance of <interfacename>BucketKeyPair</interfacename>, like <classname>SimpleBucketKeyPair</classname>.</listitem>
|
|
|
|
|
<listitem><emphasis>Using a <interfacename>Map</interfacename></emphasis> - You can pass a <interfacename>Map</interfacename> with keys for "bucket" and "key".</listitem>
|
|
|
|
|
</itemizedlist>
|
|
|
|
|
</para>
|
|
|
|
|
|
|
|
|
|
<section id="riak:template:config">
|
|
|
|
|
<title>Configuring the <classname>RiakTemplate</classname></title>
|
|
|
|
|
|
|
|
|
|
<para>This is likely the easiest path to using SDKV for Riak, as the bucket and key are passed separately. The examples that follow will assume you're using this version of the the template.</para>
|
|
|
|
|
|
|
|
|
|
<para>There are only two options you need to set to specify the Riak server to use in your <classname>RiakTemplate</classname> object: "defaultUri" and "mapReduceUri". Encoded with the URI should be placeholders for the bucket and the key, which will be filled in by the <classname>RestTemplate</classname> when the request is made.
|
|
|
|
|
|
|
|
|
|
<important><para>You can also turn the internal, ETag-based object cache off by setting <literal>useCache="false"</literal>. It's generally recommended, however, to leave the internal cache on as the ETag matching will pick up any changes made to the entry on the Riak side and your application will benefit from greatly-increased performance for often-requested objects.</para></important>
|
|
|
|
|
|
|
|
|
|
<programlisting language="xml"><![CDATA[<?xml version="1.0" encoding="UTF-8"?>
|
|
|
|
|
<beans xmlns="http://www.springframework.org/schema/beans"
|
|
|
|
|
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
|
|
|
|
xmlns:p="http://www.springframework.org/schema/p"
|
|
|
|
|
xsi:schemaLocation="http://www.springframework.org/schema/beans
|
|
|
|
|
http://www.springframework.org/schema/beans/spring-beans.xsd">
|
|
|
|
|
|
|
|
|
|
<bean id="riakTemplate" class="org.springframework.data.keyvalue.riak.core.RiakTemplate"
|
|
|
|
|
p:defaultUri="http://localhost:8098/riak/{bucket}/{key}"
|
|
|
|
|
p:mapReduceUri="http://localhost:8098/mapred"
|
|
|
|
|
p:useCache="true"/>
|
|
|
|
|
|
|
|
|
|
</beans>]]></programlisting>
|
|
|
|
|
</para>
|
|
|
|
|
|
|
|
|
|
<section id="riak:template:config:advanced">
|
|
|
|
|
<title>Advanced Template Configuration</title>
|
|
|
|
|
|
|
|
|
|
<para>There are a couple additional properties on the <classname>RiakTemplate</classname> that can be changed from their defaults. If you want to specify your own <ulink url="http://static.springsource.org/spring/docs/3.0.x/spring-framework-reference/html/validation.html#core-convert-ConversionService-API">ConversionService</ulink> to use when converting objects for storage inside Riak, then set it on the "conversionService" property:
|
|
|
|
|
|
|
|
|
|
<programlisting language="xml"><![CDATA[<?xml version="1.0" encoding="UTF-8"?>
|
|
|
|
|
<beans xmlns="http://www.springframework.org/schema/beans"
|
|
|
|
|
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
|
|
|
|
xmlns:p="http://www.springframework.org/schema/p"
|
|
|
|
|
xsi:schemaLocation="http://www.springframework.org/schema/beans
|
|
|
|
|
http://www.springframework.org/schema/beans/spring-beans.xsd">
|
|
|
|
|
|
|
|
|
|
<bean id="conversionService" class="com.mycompany.convert.MyConversionService"/>
|
|
|
|
|
<bean id="riakTemplate" class="org.springframework.data.keyvalue.riak.core.RiakTemplate"
|
|
|
|
|
p:defaultUri="http://localhost:8098/riak/{bucket}/{key}"
|
|
|
|
|
p:mapReduceUri="http://localhost:8098/mapred"
|
|
|
|
|
p:conversionService-ref="conversionService"/>
|
|
|
|
|
|
|
|
|
|
</beans>]]></programlisting>
|
|
|
|
|
</para>
|
|
|
|
|
|
|
|
|
|
<para>Depending on the application, it might be useful to set default Quality-of-Service parameters. In Riak paralance, these are the <ulink url="https://wiki.basho.com/display/RIAK/REST+API#RESTAPI-Setbucketproperties">"dw", "w", and "r" parameters</ulink>. They can be set to an integer representing the number of vnodes that need to report having received the data before declaring the operation a success, or the string "one", "all", or (the default) "quorum". These values can be overridden by passing a different set of <interfacename>QosParameters</interfacename> to the set/get operation you're performing.
|
|
|
|
|
|
|
|
|
|
<programlisting language="xml"><![CDATA[<?xml version="1.0" encoding="UTF-8"?>
|
|
|
|
|
<beans xmlns="http://www.springframework.org/schema/beans"
|
|
|
|
|
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
|
|
|
|
xmlns:p="http://www.springframework.org/schema/p"
|
|
|
|
|
xsi:schemaLocation="http://www.springframework.org/schema/beans
|
|
|
|
|
http://www.springframework.org/schema/beans/spring-beans.xsd">
|
|
|
|
|
|
|
|
|
|
<bean id="qos" class="org.springframework.data.keyvalue.riak.core.RiakQosParameters"
|
|
|
|
|
p:durableWriteThreshold="all"
|
|
|
|
|
p:writeThreshold="all"/>
|
|
|
|
|
<bean id="riakTemplate" class="org.springframework.data.keyvalue.riak.core.RiakTemplate"
|
|
|
|
|
p:defaultUri="http://localhost:8098/riak/{bucket}/{key}"
|
|
|
|
|
p:mapReduceUri="http://localhost:8098/mapred"
|
|
|
|
|
p:defaultQosParameters-ref="qos"/>
|
|
|
|
|
|
|
|
|
|
</beans>]]></programlisting>
|
|
|
|
|
</para>
|
|
|
|
|
|
|
|
|
|
<para>You can also set a specific <literal>ClassLoader</literal> to use when loading objects from Riak. Just set the <literal>classLoader</literal> property:
|
|
|
|
|
|
|
|
|
|
<programlisting language="xml"><![CDATA[<?xml version="1.0" encoding="UTF-8"?>
|
|
|
|
|
<beans xmlns="http://www.springframework.org/schema/beans"
|
|
|
|
|
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
|
|
|
|
xmlns:p="http://www.springframework.org/schema/p"
|
|
|
|
|
xsi:schemaLocation="http://www.springframework.org/schema/beans
|
|
|
|
|
http://www.springframework.org/schema/beans/spring-beans.xsd">
|
|
|
|
|
|
|
|
|
|
<bean id="riakTemplate" class="org.springframework.data.keyvalue.riak.core.RiakTemplate"
|
|
|
|
|
p:defaultUri="http://localhost:8098/riak/{bucket}/{key}"
|
|
|
|
|
p:mapReduceUri="http://localhost:8098/mapred"
|
|
|
|
|
p:classLoader-ref="customClassLoader"/>
|
|
|
|
|
|
|
|
|
|
</beans>]]></programlisting>
|
|
|
|
|
|
|
|
|
|
</para>
|
|
|
|
|
|
|
|
|
|
</section>
|
|
|
|
|
</section>
|
|
|
|
|
|
|
|
|
|
<section id="riak:template:objects">
|
|
|
|
|
<title>Working with Objects using the <classname>RiakTemplate</classname></title>
|
|
|
|
|
|
|
|
|
|
<para>One of the primary goals of the SDKV project is to make accessing Key/Value stores easier for the developer by taking away the mundane tasks of basic IO, buffering, type conversion, exception handling, and sundry other logistical concerns so the developer can focus on creating great applications. SDKV for Riak works toward this goal by making basic persistence and data access as easy as using a <classname>Map</classname>.</para>
|
|
|
|
|
|
|
|
|
|
<section id="riak:template:objects:set">
|
|
|
|
|
<title>Saving data into Riak</title>
|
|
|
|
|
|
|
|
|
|
<para>To store data in Riak, use one of the six different <literal>set</literal> methods:
|
|
|
|
|
<programlisting language="java"><![CDATA[import org.springframework.data.keyvalue.riak.core.RiakTemplate;
|
|
|
|
|
|
|
|
|
|
public class Example {
|
|
|
|
|
|
|
|
|
|
@Autowired
|
|
|
|
|
RiakTemplate riak;
|
|
|
|
|
|
|
|
|
|
public void setData(String bucket, String key, String data) throws Exception {
|
|
|
|
|
riak.set(bucket, key, data); // Set as Content-Type: text/plain
|
|
|
|
|
riak.setAsBytes(bucket, key, data.getBytes()); // Set as Content-Type: application/octet-stream
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
public void setData(String bucket, String key, MyPojo data) throws Exception {
|
|
|
|
|
riak.set(bucket, key, data); // Converted to JSON automatically, Content-Type: application/json
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
}
|
|
|
|
|
]]></programlisting>
|
|
|
|
|
</para>
|
|
|
|
|
|
|
|
|
|
<para>Additionally, there is a <literal>setWithMetaData</literal> method that takes a <classname>Map</classname> of metadata that will be set as the outgoing HTTP headers. To set <ulink url="https://wiki.basho.com/display/RIAK/REST+API#RESTAPI-Storeaneworexistingobjectwithakey">custom metadata</ulink>, your key should be prefixed with <literal>X-Riak-Meta-</literal> e.g. <literal>X-Riak-Meta-Custom-Header</literal>.</para>
|
|
|
|
|
|
|
|
|
|
<section id="riak:template:objects:set:genid">
|
|
|
|
|
<title>Letting Riak generate the key</title>
|
|
|
|
|
|
|
|
|
|
<para>Riak has the ability to generate random IDs for you when storing objects. The <classname>RiakTemplate</classname> exposes this capability via the <literal>put</literal> method. It will return the ID it generated for you as a <classname>String</classname>.
|
|
|
|
|
<programlisting language="java"><![CDATA[import org.springframework.data.keyvalue.riak.core.RiakTemplate;
|
|
|
|
|
|
|
|
|
|
public class Example {
|
|
|
|
|
|
|
|
|
|
@Autowired
|
|
|
|
|
RiakTemplate riak;
|
|
|
|
|
|
|
|
|
|
public String setData(String bucket, String data) throws Exception {
|
|
|
|
|
String id = riak.put(bucket, data); // Returns the generated ID
|
|
|
|
|
return id;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
}
|
|
|
|
|
]]></programlisting>
|
|
|
|
|
</para>
|
|
|
|
|
</section>
|
|
|
|
|
</section>
|
|
|
|
|
|
|
|
|
|
<section id="riak:template:objects:get">
|
|
|
|
|
<title>Retrieving data from Riak</title>
|
|
|
|
|
|
|
|
|
|
<para>Retrieving data from Riak is just as easy. There are actually 13 different <literal>get</literal> methods on <classname>RiakTemplate</classname> that give the developer a wide range options for accessing and converting your data.</para>
|
|
|
|
|
|
|
|
|
|
<para>Assuming you've stored a POJO using an appropriate <literal>set</literal> method, you can retrieve that object from Riak using a <literal>get</literal>:
|
|
|
|
|
<programlisting language="java"><![CDATA[import org.springframework.data.keyvalue.riak.core.RiakTemplate;
|
|
|
|
|
|
|
|
|
|
public class Example {
|
|
|
|
|
|
|
|
|
|
@Autowired
|
|
|
|
|
RiakTemplate riak;
|
|
|
|
|
|
|
|
|
|
public void getData(String bucket, String key) throws Exception {
|
|
|
|
|
// What you get depends on Content-Type.
|
|
|
|
|
// application/json=Map, text/plain=String, etc...
|
|
|
|
|
Object o = riak.get(bucket, key);
|
|
|
|
|
|
|
|
|
|
// If your entry is Content-Type: application/json...
|
|
|
|
|
// It will automatically be converted when retrieved.
|
|
|
|
|
MyPojo s = riak.getAsType(bucket, key, MyPojo.class);
|
|
|
|
|
|
|
|
|
|
// If your entry is Content-Type: application/octet-stream,
|
|
|
|
|
// you can access the raw bytes.
|
|
|
|
|
byte[] b = riak.getAsBytes(bucket, key);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
}
|
|
|
|
|
]]></programlisting>
|
|
|
|
|
</para>
|
|
|
|
|
</section>
|
|
|
|
|
</section>
|
|
|
|
|
|
|
|
|
|
<section id="riak:links">
|
|
|
|
|
<title>Linking Entries</title>
|
|
|
|
|
|
|
|
|
|
<para>Riak has the ability to <ulink url="https://wiki.basho.com/display/RIAK/Links">link entries together using an arbitrary tag</ulink>. This relationship information is stored in the <literal>Link</literal> header. The <classname>RiakTemplate</classname> exposes a method for linking entries together called <ulink url="../../api/org/springframework/data/keyvalue/riak/core/RiakTemplate.html#link(B1, K1, B2, K2, java.lang.String)"><literal>link</literal></ulink>. Its usage is quite simple:
|
|
|
|
|
|
|
|
|
|
<important><para>A link is uni-directional, so keep in mind that the bucket and key you pass first should be that of the child (or target) object and the second set of bucket/key pairs you pass to the <literal>link</literal> method is the source of the relationship. It's this second entry that will receive an updated <literal>Link</literal> header that points to the child or target entry.</para></important>
|
|
|
|
|
|
|
|
|
|
<programlisting language="java"><![CDATA[
|
|
|
|
|
@Autowired
|
|
|
|
|
RiakTemplate riak;
|
|
|
|
|
|
|
|
|
|
riak.link("childbucket", "childkey", "sourcebucket", "sourcekey", "tagname");
|
|
|
|
|
]]></programlisting>
|
|
|
|
|
|
|
|
|
|
<para>Now, querying the metadata on the entry at <literal>sourcebucket:sourcekey</literal> will result in a Link header that points to the child object: <literal></riak/childbucket/childkey>; riaktag="tagname"</literal></para>
|
|
|
|
|
|
|
|
|
|
</para>
|
|
|
|
|
|
|
|
|
|
<section id="riak:links:walking">
|
|
|
|
|
<title>Link Walking</title>
|
|
|
|
|
|
|
|
|
|
<para>When entries are linked together in Riak, those relationships can be efficiently traversed on the server using a feature called <ulink url="http://blog.basho.com/2010/02/24/link-walking-by-example/">Link Walking</ulink>. Rather than requesting each object in a link's relationship individually, a link walk pulls all the related objects at once and sends that data back to the client as MIME-encoded multipart data. As such, it requires special processing to convert those multiple entries into a <interfacename>List</interfacename> of objects, just as if you had used a <literal>get</literal> method. If you don't specify a type to convert the objects to, the <ulink url="../../api/org/springframework/data/keyvalue/riak/core/RiakTemplate.html#linkWalk(B, K, java.lang.String)"><literal>linkWalk</literal></ulink> method will try to infer it from the bucket name. If the bucket name is not a valid class name, it will default to using a <interfacename>java.util.Map</interfacename>.</para>
|
|
|
|
|
|
|
|
|
|
<para>To link walk a relationship and return a list of custom POJOs, you would do something like this:
|
|
|
|
|
<programlisting language="java"><![CDATA[
|
|
|
|
|
@Autowired
|
|
|
|
|
RiakTemplate riak;
|
|
|
|
|
|
|
|
|
|
List<MyPojo> result = riak.linkWalk("sourcebucket", "sourcekey", "tagname", MyPojo.class);
|
|
|
|
|
]]></programlisting>
|
|
|
|
|
</para>
|
|
|
|
|
</section>
|
|
|
|
|
</section>
|
|
|
|
|
|
|
|
|
|
<section id="riak:mapred">
|
|
|
|
|
<title>Map/Reduce</title>
|
|
|
|
|
|
|
|
|
|
<para>Riak supports Map/Reduce functionality in a couple different ways. You can specify the Javascript source to execute (termed "anonymous" Javascript), you can reference some Javascript already stored in Riak at a specfic bucket and key, or you can reference an Erlang module and function. The Map/Reduce support in SDKV covers all these bases by giving you meaningful abstractions over the Map/Reduce job that represent the various aspects of the Map/Reduce process.</para>
|
|
|
|
|
|
|
|
|
|
<para>At the highest level, every Map/Reduce request is represented by a <ulink url="../../api/org/springframework/data/keyvalue/riak/mapreduce/MapReduceJob.html">MapReduceJob</ulink>. The <literal>MapReduceJob</literal> represents the <literal>inputs</literal>, the <literal>phases</literal>, and the optional <literal>arg</literal> to send to Riak to execute the Map/Reduce job. The <literal>toJson</literal> method is responsible for serializing the entire job into the appropriate JSON data to send to Riak.</para>
|
|
|
|
|
|
|
|
|
|
<section id="riak:mapred:inputs">
|
|
|
|
|
<title>Specifying Inputs</title>
|
|
|
|
|
|
|
|
|
|
<para>Riak will accept either a string denoting the bucket in which to get the list of keys to operate on, or a <interfacename>List</interfacename> of <interfacename>List</interfacename>s denoting the bucket/key pairs to operate on while executing this Map/Reduce job. If you call the <literal>addInputs</literal> method on the job passing a <interfacename>List</interfacename> with a single string entry, the job will assume you want to operate on an entire bucket. Otherwise, you'll need to pass a multi-dimensional <interfacename>List</interfacename> of bucket/key pairs.</para>
|
|
|
|
|
|
|
|
|
|
<para>To operate on an entire bucket:
|
|
|
|
|
<programlisting language="java"><![CDATA[
|
|
|
|
|
@Autowired
|
|
|
|
|
RiakTemplate riak;
|
|
|
|
|
|
|
|
|
|
RiakMapReduceJob job = riak.createMapReduceJob();
|
|
|
|
|
List<String> bucket = new ArrayList<String>() {{
|
|
|
|
|
add("mybucket");
|
|
|
|
|
}};
|
|
|
|
|
job.addInputs(bucket); // Will M/R entire bucket
|
|
|
|
|
]]></programlisting>
|
|
|
|
|
</para>
|
|
|
|
|
|
|
|
|
|
<para>To operate on a set of keys:
|
|
|
|
|
<programlisting language="java"><![CDATA[import org.springframework.data.keyvalue.riak.mapreduce.*;
|
|
|
|
|
|
|
|
|
|
@Autowired
|
|
|
|
|
RiakTemplate riak;
|
|
|
|
|
|
|
|
|
|
RiakMapReduceJob job = riak.createMapReduceJob();
|
|
|
|
|
|
|
|
|
|
List<String> pair = new ArrayList<String>() {{
|
|
|
|
|
add("mybucket");
|
|
|
|
|
add("mykey");
|
|
|
|
|
}};
|
|
|
|
|
List<List<String>> keys = new ArrayList<List<String>>() {{
|
|
|
|
|
add(pair);
|
|
|
|
|
}};
|
|
|
|
|
job.addInputs(keys); // Will M/R only specified keys
|
|
|
|
|
]]></programlisting>
|
|
|
|
|
</para>
|
|
|
|
|
</section>
|
|
|
|
|
|
|
|
|
|
<section id="riak:mapred:phases">
|
|
|
|
|
<title>Defining Phases</title>
|
|
|
|
|
|
|
|
|
|
<para>Map/Reduce operations in Riak are broken up into phases. Phases contain a <ulink url="../../api/org/springframework/data/keyvalue/riak/mapreduce/MapReduceOperation.html"><interfacename>MapReduceOperation</interfacename></ulink>. There are currently two implementations to handle Javascript or Erlang M/R operations: <ulink url="../../api/org/springframework/data/keyvalue/riak/mapreduce/JavascriptMapReduceOperation.html"><classname>JavascriptMapReduceOperation</classname></ulink> and <ulink url="../../api/org/springframework/data/keyvalue/riak/mapreduce/ErlangMapReduceOperation.html"><classname>ErlangMapReduceOperation</classname></ulink>.</para>
|
|
|
|
|
|
|
|
|
|
<para>An example Map/Reduce job defining a single "map" phase defined in anonymous Javascript might look like this:
|
|
|
|
|
<programlisting language="java"><![CDATA[import org.springframework.data.keyvalue.riak.mapreduce.*;
|
|
|
|
|
|
|
|
|
|
@Autowired
|
|
|
|
|
RiakTemplate riak;
|
|
|
|
|
|
|
|
|
|
RiakMapReduceJob job = riak.createMapReduceJob();
|
|
|
|
|
List<String> bucket = new ArrayList<String>() {{
|
|
|
|
|
add("mybucket");
|
|
|
|
|
}};
|
|
|
|
|
|
|
|
|
|
job.addInputs(bucket); // M/R the entire bucket
|
|
|
|
|
|
|
|
|
|
MapReduceOperation mapOper = new JavascriptMapReduceOperation("function(v){ ...M/R function body... }");
|
|
|
|
|
MapReducePhase mapPhase = new RiakMapReducePhase("map", "javascript", mapOper);
|
|
|
|
|
|
|
|
|
|
job.addPhase(mapPhase);
|
|
|
|
|
]]></programlisting>
|
|
|
|
|
</para>
|
|
|
|
|
</section>
|
|
|
|
|
|
|
|
|
|
<section id="riak:mapred:exec">
|
|
|
|
|
<title>Executing and Working with the Result</title>
|
|
|
|
|
|
|
|
|
|
<para>To execute a configured job on your Riak server, use either the synchronous <literal>execute</literal> or asynchronous <literal>submit</literal> methods of your configured <classname>RiakTemplate</classname>:
|
|
|
|
|
|
|
|
|
|
<programlisting language="java"><![CDATA[
|
|
|
|
|
Object o = riak.execute(job); // Results of last Map or Reduce phase. Should be a List<?>
|
|
|
|
|
|
|
|
|
|
...or...
|
|
|
|
|
|
|
|
|
|
List<MyPojo> o = riak.execute(job, MyPojo.class); // Coerce to given type
|
|
|
|
|
|
|
|
|
|
...or...
|
|
|
|
|
|
|
|
|
|
Future<List<?>> f = riak.submit(job); // Job runs in a separate thread
|
|
|
|
|
]]></programlisting>
|
|
|
|
|
</para>
|
|
|
|
|
</section>
|
|
|
|
|
</section>
|
|
|
|
|
|
|
|
|
|
<section id="riak:buckets">
|
|
|
|
|
<title>Managing Bucket Properties</title>
|
|
|
|
|
|
|
|
|
|
<para>It's sometimes useful to manage settings like the Quality-of-Service parameters <literal>w</literal> and <literal>dw</literal> (write and durable write thresholds) and the <literal>n_val</literal> setting at the bucket level. It's also possible to list the keys in a particular bucket by calling the <literal>getBucketSchema</literal> method, passing <literal>true</literal> as the second parameter, which tells the <classname>RiakTemplate</classname> to list the keys.</para>
|
|
|
|
|
|
|
|
|
|
<para>To list the keys in a bucket, you would do something like this:
|
|
|
|
|
|
|
|
|
|
<programlisting language="java"><![CDATA[
|
|
|
|
|
@Autowired
|
|
|
|
|
RiakTemplate riak;
|
|
|
|
|
|
|
|
|
|
Map<String, Object> schema = riak.getBucketSchema("mybucket", true);
|
|
|
|
|
List<String> keys = schema.get("keys")
|
|
|
|
|
for(String key : keys) {
|
|
|
|
|
...do something with each key...
|
|
|
|
|
}
|
|
|
|
|
]]></programlisting>
|
|
|
|
|
</para>
|
|
|
|
|
|
|
|
|
|
<para>To update the bucket settings, pass a <interfacename>Map</interfacename> of properties:
|
|
|
|
|
|
|
|
|
|
<programlisting language="java"><![CDATA[
|
|
|
|
|
@Autowired
|
|
|
|
|
RiakTemplate riak;
|
|
|
|
|
|
|
|
|
|
Map<String, Integer> props = new HashMap<String, Integer>();
|
|
|
|
|
props.put("n_val", 6);
|
|
|
|
|
props.put("dw", 3);
|
|
|
|
|
|
|
|
|
|
riak.updateBucketSchema("mybucket", props);
|
|
|
|
|
]]></programlisting>
|
|
|
|
|
|
|
|
|
|
<para>Only the properties specified in the passed-in <interfacename>Map</interfacename> will be updated. Properties that have already been set in previous operations and not specified in this operation will be unaffected.</para>
|
|
|
|
|
</para>
|
|
|
|
|
|
|
|
|
|
</section>
|
|
|
|
|
|
|
|
|
|
<section id="riak:async">
|
|
|
|
|
<title>Asynchronous Access</title>
|
|
|
|
|
|
|
|
|
|
<para>SDKV for Riak also includes an asynchronous version of most of the methods available to the <classname>RiakTemplate</classname>, whose method calls are all synchronous. The asynchronous version of the template is called <classname>AsyncRiakTemplate</classname>.</para>
|
|
|
|
|
|
|
|
|
|
<section id="riak:async:config">
|
|
|
|
|
<title>Template Configuration</title>
|
|
|
|
|
|
|
|
|
|
<para>The <classname>AsyncRiakTemplate</classname> has the same basic configuration properties as the synchronous <classname>RiakTemplate</classname>. The only other property specific to the <classname>AsyncRiakTemplate</classname> you might want to configure is the thread pool the template uses to execute tasks asynchronously (by default a cached <ulink url="http://download.oracle.com/javase/6/docs/api/java/util/concurrent/ThreadPoolExecutor.html">ThreadPoolExecutor</ulink>). Set your <classname>ExecutorService</classname> on the template's <literal>workerPool</literal> property.</para>
|
|
|
|
|
|
|
|
|
|
</section>
|
|
|
|
|
|
|
|
|
|
<section id="riak:async:callbacks">
|
|
|
|
|
<title>Callbacks</title>
|
|
|
|
|
|
|
|
|
|
<para>Using the asynchronous Riak support in SDKV means you'll be relying on callbacks to execute your business logic when the requested operation is completed. All asynchronous operations follow a similar pattern:
|
|
|
|
|
<itemizedlist>
|
|
|
|
|
<listitem>They are named similarly to their synchronous counterparts.</listitem>
|
|
|
|
|
<listitem>They take a <interfacename>AsyncKeyValueStoreOperation<?, ?></interfacename> as a final parameter.</listitem>
|
|
|
|
|
<listitem>They return a <interfacename>Future<?></interfacename>.</listitem>
|
|
|
|
|
</itemizedlist>
|
|
|
|
|
</para>
|
|
|
|
|
|
|
|
|
|
<para>To perform an asynchronous <literal>get</literal> on a JSON-serialized <interfacename>Map</interfacename> object which returns a custom object from the callback, you'd do something like:
|
|
|
|
|
<programlisting language="java"><![CDATA[@Autowired
|
|
|
|
|
AsyncRiakTemplate riak;
|
|
|
|
|
|
|
|
|
|
Future<MyObject> future = riak.get("mybucket", "mykey", new AsyncKeyValueStoreOperation<Map, MyObject>() {
|
|
|
|
|
|
|
|
|
|
MyObject obj = new MyObject();
|
|
|
|
|
|
|
|
|
|
MyObject completed(KeyValueStoreMetaData meta, Map result) {
|
|
|
|
|
obj.setName(result.get("name"));
|
|
|
|
|
return obj;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
MyObject failed(Throwable error) {
|
|
|
|
|
obj.setError(error);
|
|
|
|
|
return obj;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
});
|
|
|
|
|
|
|
|
|
|
// Maybe do other work while waiting...
|
|
|
|
|
MyObject obj = future.get();
|
|
|
|
|
]]></programlisting>
|
|
|
|
|
</para>
|
|
|
|
|
</section>
|
|
|
|
|
|
|
|
|
|
</section>
|
|
|
|
|
|
|
|
|
|
<section id="riak:groovy">
|
|
|
|
|
<title>Groovy Builder Support</title>
|
|
|
|
|
|
|
|
|
|
<para>If your application uses Groovy, either in a standalone context, or as part of a Grails application, then you could benefit from using the Groovy <ulink url="../../api/org/springframework/data/keyvalue/riak/groovy/RiakBuilder.html"><classname>RiakBuilder</classname></ulink> that comes with SDKV for Riak. Underneath, it uses the <classname>AsyncRiakTemplate</classname>. To use the <classname>RiakBuilder</classname>, pass the constructor a configured <classname>AsyncRiakTemplate</classname>.</para>
|
|
|
|
|
|
|
|
|
|
<important><para>Instances of <classname>RiakBuilder</classname> are NOT thread-safe and should not be shared across threads.</para></important>
|
|
|
|
|
|
|
|
|
|
<para>The <classname>RiakBuilder</classname> implements an easy-to-use DSL for interacting with Riak. It doesn't implement the full set of methods available on the underlying <classname>AsyncRiakTemplate</classname> but a subset. The methods that the <classname>RiakBuilder</classname> responds to are:
|
|
|
|
|
<itemizedlist>
|
|
|
|
|
<listitem>set</listitem>
|
|
|
|
|
<listitem>setAsBytes</listitem>
|
|
|
|
|
<listitem>put</listitem>
|
|
|
|
|
<listitem>get</listitem>
|
|
|
|
|
<listitem>getAsBytes</listitem>
|
|
|
|
|
<listitem>getAsType</listitem>
|
|
|
|
|
<listitem>containsKey</listitem>
|
|
|
|
|
<listitem>delete</listitem>
|
|
|
|
|
<listitem>foreach</listitem>
|
|
|
|
|
</itemizedlist>
|
|
|
|
|
</para>
|
|
|
|
|
|
|
|
|
|
<section id="riak:groovy:dsl">
|
|
|
|
|
<title>Riak DSL Usage</title>
|
|
|
|
|
|
|
|
|
|
<para>The following example illustrates the different uses of the Riak DSL, including batching requests together into a logical group, using a default bucket name (the node directly beneath <literal>riak</literal> will be considered the default bucket to use for the contained operations unless a different one is specified on the operation itself):
|
|
|
|
|
<programlisting language="java"><![CDATA[
|
|
|
|
|
def riak = new RiakBuilder(asyncRiakTemplate)
|
|
|
|
|
riak {
|
|
|
|
|
test {
|
|
|
|
|
put(value: [test: "value"]) { completed { v, meta -> meta.key }}
|
|
|
|
|
put(value: [test: "value"]) { completed { v, meta -> meta.key }}
|
|
|
|
|
put(value: [test: "value"]) { completed { v, meta -> meta.key }}
|
|
|
|
|
put(value: [test: "value"]) { completed { v, meta -> meta.key }}
|
|
|
|
|
|
|
|
|
|
mapreduce {
|
|
|
|
|
query {
|
|
|
|
|
map(arg: [test: "arg", alist: [1, 2, 3, 4]]) {
|
|
|
|
|
source "function(v, keyInfo, arg){ return [1]; }"
|
|
|
|
|
}
|
|
|
|
|
reduce {
|
|
|
|
|
source "function(v){ return Riak.reduceSum(v); }"
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
failed { it.printStackTrace() }
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
def results = riak.results
|
|
|
|
|
|
|
|
|
|
riak.foreach(bucket: "test") {
|
|
|
|
|
completed { v, meta ->
|
|
|
|
|
riak.delete(bucket: "test", key: meta.key)
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
]]></programlisting>
|
|
|
|
|
</para>
|
|
|
|
|
|
|
|
|
|
<para>Some important things to note from this example:
|
|
|
|
|
<itemizedlist>
|
|
|
|
|
<listitem>Each operation in the Riak DSL has two callbacks: <literal>completed</literal> and <literal>failed</literal>.</listitem>
|
|
|
|
|
<listitem>The <literal>completed</literal> closure is passed either the result object, or, if your closure is defined with two parameters, the result object and the <ulink url="../../api/org/springframework/data/keyvalue/riak/core/RiakMetaData.html">metadata</ulink> associated with that entry.</listitem>
|
|
|
|
|
<listitem>Operations can be enclosed in an arbitrarily-named closure which the builder interprets as a default bucket name (in this case, the node "test" tells the builder to use the bucket name "test" for a default, unless one is specified on one of the enclosed operations).</listitem>
|
|
|
|
|
<listitem>Each operation within a builder's execution will be accumulated inside the special <literal>results</literal> property. Code that needs to know the output of individual operations within the batch can get access to that object through this property. Note that this means that <classname>RiakBuilder</classname> instances are NOT thread-safe.</listitem>
|
|
|
|
|
</itemizedlist>
|
|
|
|
|
</para>
|
|
|
|
|
|
|
|
|
|
<important><para>Even though the Riak DSL uses an asynchronous template underneath, all operations performed through the DSL will, by default, block until complete. To get a truly asynchronous operation, pass the parameter <literal>wait: 0</literal> (or give a meaningful timeout in milliseconds to wait for the operation to complete) on the operation.</para></important>
|
|
|
|
|
|
|
|
|
|
<section id="riak:groovy:dsl:qos">
|
|
|
|
|
<title>QosParameters on Riak DSL Operations</title>
|
|
|
|
|
|
|
|
|
|
<para>You can pass <interfacename>QosParameters</interfacename> to Riak DSL operations by simply defining them as parameters to the operation:
|
|
|
|
|
<programlisting language="java"><![CDATA[
|
|
|
|
|
def riak = new RiakBuilder(asyncRiakTemplate)
|
|
|
|
|
|
|
|
|
|
def myobj = riak.set(bucket: "mybucket", key: "mykey", qos: ["dw": "all"])
|
|
|
|
|
]]></programlisting>
|
|
|
|
|
</para>
|
|
|
|
|
</section>
|
|
|
|
|
|
|
|
|
|
<section id="riak:groovy:dsl:output">
|
|
|
|
|
<title>Working with Riak DSL Output</title>
|
|
|
|
|
|
|
|
|
|
<para>The output of DSL operations will either be passed to the configured <literal>completed</literal> callback, or be returned to the caller if no callback is specified. In the example above, the <literal>mapreduce</literal> operation has no <literal>completed</literal> closure. Therefore, the return of the reduce phase is simply passed back to the builder, which makes that output available on the special <literal>results</literal> property.</para>
|
|
|
|
|
|
|
|
|
|
<para>To gain access to the operation's results immediately, simply assign it to a variable:
|
|
|
|
|
<programlisting language="java"><![CDATA[
|
|
|
|
|
def riak = new RiakBuilder(asyncRiakTemplate)
|
|
|
|
|
|
|
|
|
|
def myobj = riak.get(bucket: "mybucket", key: "mykey")
|
|
|
|
|
]]></programlisting>
|
|
|
|
|
|
|
|
|
|
<para>If you add a non-zero <literal>wait</literal> value to the operation, "myobj" will contain a <interfacename>Future<?></interfacename> rather than the result object itself.</para>
|
|
|
|
|
|
|
|
|
|
</para>
|
|
|
|
|
</section>
|
|
|
|
|
</section>
|
|
|
|
|
</section>
|
|
|
|
|
|
|
|
|
|
<section id="riak:io">
|
|
|
|
|
<title>Working with streams</title>
|
|
|
|
|
|
|
|
|
|
<para>SDKV for Riak includes a couple of useful helper objects to make reading and writing plain text or binary data in Riak really easy. If you want to store a file in Riak, then you can create a <classname>RiakOutputStream</classname> and simply write your data to it (making sure to call the "flush" method, which actually sends the data to Riak).
|
|
|
|
|
|
|
|
|
|
<programlisting language="java"><![CDATA[import org.springframework.data.keyvalue.riak.core.RiakTemplate;
|
|
|
|
|
import org.springframework.data.keyvalue.riak.core.io.RiakOutputStream;
|
|
|
|
|
|
|
|
|
|
public class Example {
|
|
|
|
|
|
|
|
|
|
@Autowired
|
|
|
|
|
RiakTemplate riak;
|
|
|
|
|
|
|
|
|
|
public void writeToRiak(String bucket, String key, String data) throws Exception {
|
|
|
|
|
OutputStream out = new RiakOutputStream(riak, bucket, key);
|
|
|
|
|
try {
|
|
|
|
|
out.write(data.getBytes());
|
|
|
|
|
} finally {
|
|
|
|
|
out.flush();
|
|
|
|
|
out.close();
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
}
|
|
|
|
|
]]></programlisting>
|
|
|
|
|
|
|
|
|
|
<para>Reading data from Riak is similarly easy. SDKV provides a <literal>java.io.File</literal> subclass that represents a resource in Riak. There's also a Spring IO Resource abstraction called <classname>RiakResource</classname> that can be used anywhere a <ulink url="http://static.springsource.org/spring/docs/3.0.x/spring-framework-reference/html/resources.html"><interfacename>Resource</interfacename></ulink> is required. There's also an <classname>InputStream</classname> implementation called <classname>RiakInputStream</classname>.</para>
|
|
|
|
|
|
|
|
|
|
<programlisting language="java"><![CDATA[import org.springframework.data.keyvalue.riak.core.RiakTemplate;
|
|
|
|
|
import org.springframework.data.keyvalue.riak.core.io.RiakInputStream;
|
|
|
|
|
|
|
|
|
|
public class Example {
|
|
|
|
|
|
|
|
|
|
@Autowired
|
|
|
|
|
RiakTemplate riak;
|
|
|
|
|
|
|
|
|
|
public String readFromRiak(String bucket, String key) throws Exception {
|
|
|
|
|
InputStream in = new RiakInputStream(riak, bucket, key);
|
|
|
|
|
String data;
|
|
|
|
|
...read data and work with it...
|
|
|
|
|
return data;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
}
|
|
|
|
|
]]></programlisting>
|
|
|
|
|
|
|
|
|
|
</para>
|
|
|
|
|
</section>
|
|
|
|
|
</chapter>
|
Costin Leau