0

我正在运行以下 Riak 地图阶段：

-module(delete_map_function).

-export([get_keys/3]).

%Returns bucket and key pairs from a map phase
get_keys(Value,_Keydata,_Arg) ->
  [[riak_object:bucket(Value),riak_object:key(Value)]].

以及以下 Riak 减少阶段：http ://contrib.basho.com/delete_keys.html

我不断收到此错误消息：

{"phase":0,"error":"function_clause","input":"{{error,notfound},{<<\"my_bucket\">>,<<\"item_key\">>},undefined}","type":"error","stack":"[{riak_object,bucket,[{error,notfound}],[{file,\"src/riak_object.erl\"},{line,251}]},{delete_map_function,get_keys,3,[{file,\"delete_map_function.erl\"},{line,7}]},{riak_kv_mrc_map,map,3,[{file,\"src/riak_kv_mrc_map.erl\"},{line,164}]},{riak_kv_mrc_map,process,3,[{file,\"src/riak_kv_mrc_map.erl\"},{line,140}]},{riak_pipe_vnode_worker,process_input,3,[{file,\"src/riak_pipe_vnode_worker.erl\"},{line,444}]},{riak_pipe_vnode_worker,wait_for_input,2,[{file,\"src/riak_pipe_vnode_worker.erl\"},{line,376}]},{gen_fsm,...},...]"}

我正在通过 Java 运行这项工作：

MapReduceResult mapReduceResult = RiakUtils.getPBClient().mapReduce(iq)
                 .addMapPhase(new NamedErlangFunction("delete_map_function", "get_keys"))
                 .addReducePhase(new NamedErlangFunction("delete_reduce_function", "delete"))
                 .execute();

我在某处读到我应该filter_notfound在 Map 阶段使用该参数，但即使在添加它之后我仍然会收到错误：

MapReduceResult mapReduceResult = RiakUtils.getPBClient().mapReduce(iq)
                 .addMapPhase(new NamedErlangFunction("delete_map_function", "get_keys"), "filter_notfound")
                 .addReducePhase(new NamedErlangFunction("delete_reduce_function", "delete"))
                 .execute();

我正在运行 Riak 1.3 并使用Riak Java 客户端v1.1.0

4

1 回答 1

1

第一的。我认为通过 map/reduce 阶段删除键不是有效的方法。如果您有键列表并将其提供给映射阶段 riak 将首先读取所有对象，然后将其提供给您的函数。所以如果你只需要删除对象，最好不要读取。

第二。你所有的 map/reduce 函数都应该写成以下例外：

  • 而不是，由于 riak 的最终性质Value，您可以获得。{error, notfound}
  • 您也可以将已删除的 riak 对象作为Value. 您可以知道该对象已被特殊标志删除dict:is_key(<<"X-Riak-Deleted">>, riak_object:get_metadata(RiakObj))：

第三。要修复您的错误，您应该从列表中过滤 notfound 键：

get_keys({error, notfound},_Keydata,_Arg) ->
    [];
get_keys(Value,_Keydata,_Arg) ->
    [[riak_object:bucket(Value),riak_object:key(Value)]].
于 2013-03-24T07:10:45.387 回答