MapReduce riak

1、今天看了一下riak的mapred/3/4/5方法，以下几处要留意的地方。

 

 

mapred(Pid::pid(), Inputs::mapred_inputs(), Query::[mapred_queryterm()]) -> {ok, mapred_result()} | {error, {badqterm, mapred_queryterm()}} | {error, timeout} | {error, term()}

Equivalent to mapred(Inputs, Query, default_timeout(mapred)).

 

mapred_inputs()这个比较好理解

mapred_inputs() = [{bucket(), key()} | {bucket(), key(), term()}] | {modfun, Module::atom(), Function::atom(), [term()]} | bucket() | {index, bucket(), Index::binary(), key()} | {index,bucket(), Index::binary(), StartKey::key(), EndKey::key()}

 

mapred_queryterm()

mapred_queryterm() = {map, mapred_funterm(), Arg::term(), Accumulate::boolean()} | {reduce, mapred_funterm(), Arg::term(), Accumulate::boolean()} | {link, Bucket::riakc_obj:bucket(), Tag::term(), Accumulate::boolean()}

这个参数理解需要花些 时间

{map, mapred_funterm(), Arg::term(), Accumulate::boolean()} 

 

mapred_funterm()

mapred_funterm() = {modfun, Module::atom(), Function::atom()} | {qfun, function()} | {strfun, list() | binary()}

以这个简单说明{qfun, function()}

 

 

 Count = fun(G, undefined, none) ->
             [dict:from_list([{I, 1}
              || I <- binary_to_term(riak_object:get_value(G))])]
           end.

 

    此处的undefined 实际是输入参数{bucket(), key(), term()}中的term，默认是undefined

     此处的none 实际是query参数[{map, {qfun, Count}, none, false},中的none

 

1> {ok, Client} = riakc_pb_socket:start("127.0.0.1", 8087).
2> Mine = riakc_obj:new(<<"groceries">>, <<"mine">>,
                        term_to_binary(["eggs", "bacon"])).
3> Yours = riakc_obj:new(<<"groceries">>, <<"yours">>,
                         term_to_binary(["bread", "bacon"])).
4> riakc_pb_socket:put(Client, Yours, [{w, 1}]).
5> riakc_pb_socket:put(Client, Mine, [{w, 1}]).

Now that we have a client and some data, let’s run a query and count how many occurances of groceries.

6> Count = fun(G, undefined, none) ->
             [dict:from_list([{I, 1}
              || I <- binary_to_term(riak_object:get_value(G))])]
           end.
7> Merge = fun(Gcounts, none) ->
             [lists:foldl(fun(G, Acc) ->
                            dict:merge(fun(_, X, Y) -> X+Y end