Riak 的 MapReduce 功能允许开发者对分布式存储中的数据执行批量扫描、过滤、转换与聚合。和单条键值读取不同，MapReduce 作业会把计算任务推送到数据所在的节点，减少跨节点数据传输，并利用集群的并行能力完成批量处理。理解其执行流程与限制，是避免批量任务影响在线业务的关键。
Riak MapReduce 的执行模型与阶段
Riak 中的 MapReduce 作业由多个阶段组成，常见的阶段包括 map、reduce 和 link。map 阶段接收一个键值对象作为输入，经过用户自定义的逻辑处理后返回零个、一个或多个中间结果。这些中间结果会被自动收集起来，成为下一个阶段的输入。如果存在 reduce 阶段，则会把上游阶段产生的所有结果作为一个列表传入，由归约函数完成聚合计算。
这种设计使得批量处理能够充分利用 Riak 集群的分布式特性。当输入是一个 bucket 时，系统会根据数据分布将 key 列表分发到各个节点，节点在本地执行 map 函数，不需要把原始对象全部拉到单一协调节点。只有在进入 reduce 阶段时，中间结果才会被汇聚到执行归约的节点上。因此，合理的 map 输出设计能够显著降低网络传输量和内存占用。
除了 JavaScript 之外，Riak 还支持使用 Erlang 编写 MapReduce 函数。JavaScript 版本更适合快速开发和调试，而 Erlang 原生函数通常会带来更高的执行效率，因为省去了跨语言调用的开销。不过 Erlang 函数调试难度更大，一般只在性能敏感的场景中使用。
提交批量 MapReduce 作业的实践
提交一个 Riak MapReduce 作业最直接的方式是使用 HTTP API。请求体是一个 JSON 对象，其中 inputs 字段用来指定输入范围，query 字段则定义了阶段链。下面这个例子会对名为 orders 的 bucket 中的所有对象执行一次简单的计数归约。
{
"inputs": "orders",
"query": [
{
"map": {
"language": "javascript",
"source": "function(value) { return [Riak.mapValuesJson(value)[0]]; }"
}
},
{
"reduce": {
"language": "javascript",
"source": "function(values) { return [values.length]; }"
}
}
],
"timeout": 60000
}
上面的 map 函数先将 Riak 对象解析为 JSON，然后原样返回，reduce 函数统计中间结果的数量。如果要进行更复杂的业务处理，比如筛选出特定分类的数据并汇总金额，可以把函数写在独立的 JavaScript 文件中，或者直接嵌入 source 字段。下面的 JavaScript 代码展示了如何过滤出 sale 分类并按区域汇总金额。
function map(value, keydata, arg) {
var data = Riak.mapValuesJson(value)[0];
var result = [];
if (data.category === 'sale') {
result.push(data);
}
return result;
}
function reduce(values) {
var summary = {};
for (var i = 0; i < values.length; i++) {
var item = values[i];
if (summary[item.region]) {
summary[item.region] += item.amount;
} else {
summary[item.region] = item.amount;
}
}
return [summary];
}
使用命令行工具 curl 提交时，需要将 JSON 请求体放在 -d 参数中，并设置正确的 Content-Type 头。Riak 通常监听 8098 端口，提交地址为 /mapred。虽然 HTTP 接口简单易用，但在生产环境中更推荐通过客户端库或 Protocol Buffers 接口提交作业，以减少 JSON 序列化开销。
利用 Key Filters 限定批量处理范围
如果直接以整个 bucket 作为输入，MapReduce 作业会扫描其中的所有 key，这在数据量较大时可能非常耗时。为了避免不必要的扫描，可以使用 key filters 对输入 key 进行预筛选。key filters 是一个列表，每个元素也是一个列表，代表一个过滤条件。例如只处理以 region_ 开头的 key，可以这样写。
{
"inputs": {
"bucket": "orders",
"key_filters": [["starts_with", "region_"]]
},
"query": [
{
"map": {
"language": "javascript",
"source": "function(value) { return [value]; }"
}
}
]
}
常见的 key filters 包括 starts_with、less_than、greater_than、between 和 tokenize 等。它们会先在协调节点上对 key 集合进行筛选，然后再把筛选后的 key 分发到数据节点执行 map。不过 key filters 本身也需要读取 key 列表，因此它并不能完全消除扫描成本，只是减少了实际读取对象的数量。
对于已经通过二级索引或 Riak Data Types 组织过的数据，也可以把二级索引查询结果直接作为 MapReduce 的输入。这种方式比遍历整个 bucket 更高效，但需要提前设计好索引结构。在选择输入策略时，应当根据数据规模、查询频率和业务容忍度综合考虑。
批量处理的性能优化与常见误区
很多开发者会把 Riak 的 MapReduce 当作通用查询工具来使用，这是一个常见误区。由于 MapReduce 需要扫描大量数据并经历多个阶段，它的响应时间通常远高于直接通过 key 读取或二级索引查询。如果在线业务需要实时返回结果，应当优先考虑使用二级索引、Riak Search 或者预先计算好的聚合数据，而不是动态提交 MapReduce 作业。
性能优化首先要控制 map 阶段的输出量。如果 map 函数对每个对象都返回大量中间结果，reduce 节点可能会遇到内存压力。尽量在 map 阶段完成过滤和字段裁剪，只返回归约所需的最小数据集合。此外，为作业设置合理的 timeout 也很重要。Riak 默认的超时时间可能不足以完成大规模扫描，过短会直接中断任务，过长则可能让协调节点长时间占用资源。
另一个需要注意的问题是批量任务对集群整体性能的影响。全 bucket 扫描会让数据节点忙于读取和计算，从而影响正常的在线读写。建议将批量处理安排在业务低峰期执行，或者通过限制输入范围来缩短执行时间。如果必须定期处理大量数据，可以考虑将 MapReduce 与外部流处理框架结合，例如先通过 Riak 的批量导出机制把数据离线导出，再在外部系统中完成复杂聚合，这样能更好地隔离批量负载与在线服务。
最后还要关注函数的幂等性和异常处理。Riak 在执行 MapReduce 时可能会因为节点故障而重试某些阶段，如果函数内部存在外部副作用，比如调用其他服务或写入数据库，重试可能导致重复操作。保持 map 和 reduce 函数的无状态、幂等特性，是保证批量处理结果一致性的重要前提。