我是 Riak 的新手,所以请原谅我的误解。
我能够添加新条目并按键和索引执行查询。但是我必须实现更复杂的查询,所以我尝试使用 MapReduce。
我有一个名为 Volume
的应用程序级实体,目前只有普通字段:
public class Volume implements Comparable<Volume>, Serializable {
@RiakIndex(name = "id")
@JsonProperty("id")
private Integer id;
@RiakIndex(name = "name")
@RiakKey
private String name;
@RiakIndex(name = "created_at")
@JsonProperty("created_at")
private long createdAt;
// setters, getters....
}
以下是我向 Riak DB 添加 Volume
实例的方法:
IRiakClient riakClient = RiakFactory.httpClient();
Bucket bucket = riakClient.fetchBucket(bucketName).execute();
for (int i = 0; i < n; i++) {
int id = i;
ManagedVolume volume = new ManagedVolume();
volume.setCreatedAt(System.currentTimeMillis());
volume.setId(id);
volume.setName("volume" + i);
bucket.store(volume).execute();
}
现在我可以毫无问题地检索实例,如下所示。
Collection<String> col = backet.fetchIndex(IntIndex.named("id")).from(3).to(5).execute();
但是所有使用MapReduce
的尝试都失败了:
String str = riakClient.mapReduce(bucketName, "name: volume1")
.addMapPhase(new NamedJSFunction("Riak.mapValuesJson")).
execute().getResultRaw();
我尝试在不添加Riak.mapValuesJson
的情况下执行此操作,尝试修改查询以使用id
而不是name
并包装volume`` 带引号(
"name:\"volume\"",
"name:\'volume\'"etc),但没有任何帮助。我总是收到 HTTP 状态 500 和以下错误:
{"error":"map_reduce_error"}`
这是堆栈跟踪:
Exception in thread "main" com.basho.riak.client.RiakException: java.io.IOException: {"error":"map_reduce_error"}
at com.basho.riak.client.query.MapReduce.execute(MapReduce.java:81)
at com.infinidat.riak.TryRiak.search(TryRiak.java:288)
at com.infinidat.riak.TryRiak.main(TryRiak.java:66)
Caused by: java.io.IOException: {"error":"map_reduce_error"}
at com.basho.riak.client.raw.http.ConversionUtil.convert(ConversionUtil.java:589)
at com.basho.riak.client.raw.http.HTTPClientAdapter.mapReduce(HTTPClientAdapter.java:386)
at com.basho.riak.client.query.MapReduce.execute(MapReduce.java:79)
... 2 more
我在 Riak 的 error.log
中找到了以下记录,并在 console.log
中找到了类似的记录。
2013-07-23 19:14:12.451 [error] <0.194.0> Supervisor riak_pipe_builder_sup had child undefined started with {riak_pipe_builder,start_link,undefined} at <0.18058.4> exit with reason {{modfun,riak_search,mapred_search,[<<"VolumeBucket">>,<<"name: 1">>]},error,badarg,[{ets,lookup,[schema_table,<<"VolumeBucket">>],[]},{riak_search_config,get_schema,1,[{file,"src/riak_search_config.erl"},{line,69}]},{riak_search_client,parse_query,3,[{file,"src/riak_search_client.erl"},{line,40}]},{riak_search,parse_query,3,[{file,"src/riak_search.erl"},{line,59}]},{riak_search,mapred_search,3,[{file,"src/riak_search.erl"},{line,46}]},{riak_kv_mrc_pipe,send_inputs,3,[{file,"src/riak_kv_mrc..."},...]},...]} in context child_terminated
我相信我在这里缺少一些东西。大概是配置问题?这是一个非常简单的查询。一旦这个查询成功,我显然想继续进行更复杂的查询。
最佳答案
您的mapreduce作业指定a Riak Search query as input ,如果您没有 Riak Search enabled ,这将会失败在您的集群中。 the Java client documentation 中有一些示例。这显示了如何指定不同类型的输入。
话虽如此,Riak MapReduce 并不是被设计为实时查询工具,因此我不确定它是否是适合您想要完成的任务的工具。与直接键值查找相比,它显着增加了系统负载,因为每个请求需要大量节点/分区参与。这会导致更高的延迟,也意味着它往往无法像直接 key 访问那样扩展。
当对 Riak 和其他键值存储进行数据建模时,需要预先考虑数据访问模式和查询模式以及数据结构,这与使用关系模型有很大不同。此处提供了一些与 Riak 数据建模相关的博客文章和演示文稿:
关于java - 无法使用 riak 进行映射缩减,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/17815583/