MongoDB聚合/MapReduce实现按月份+错误类型分组错误日志

我来帮你搞定这个需求！你需要的是先按月份分组，再在每个月内按错误类型聚合日志，用MongoDB的聚合管道就能高效实现，比MapReduce更简洁易维护，先给你看最优的聚合方案：

我们需要分两步分组：先按月份+错误类型聚合同类型日志，再按月份把这些类型日志整理成你想要的对象结构。

具体的聚合代码如下：

rfidAttErrorSchema.aggregate([
  // 第一步：过滤符合条件的日志文档
  { $match: condition },
  // 第一次分组：按「月份+错误类型」组合键分组，收集同类型的所有日志
  {
    $group: {
      _id: {
        month: { $month: "$error_date" },
        errorType: "$error_type"
      },
      errorLogs: { $push: "$$ROOT" } // 把该组下的原始日志文档全部收集到数组
    }
  },
  // 第二次分组：仅按月份分组，整理每个月内的错误类型日志结构
  {
    $group: {
      _id: "$_id.month",
      logs: {
        $push: {
          k: "$_id.errorType",
          v: "$errorLogs"
        }
      }
    }
  },
  // 将logs数组转换为键值对对象（错误类型为键，日志数组为值）
  {
    $addFields: {
      logs: { $arrayToObject: "$logs" }
    }
  },
  // 按月份倒序排序（如果需要正序，改成{ _id: 1 }即可）
  { $sort: { _id: -1 } }
], function (err, data) {
  if (err) {
    return res.status(500).json({ result: err });
  }
  console.log("data is now data", data);
  res.json({ result: data });
});

代码分步解释： #

1. 第一次$group：

   • 用复合_id（月份+错误类型）确保每个组是「同一个月的同一种错误类型」
   • $push: "$$ROOT"把该组下的所有原始日志文档收集到errorLogs数组中

2. 第二次$group：

   • 仅按月份分组，把每个错误类型对应的{k: 错误类型名称, v: 日志数组}结构推送到logs数组

3. $addFields + $arrayToObject：

   • 把logs数组转换成你需要的对象格式，完美匹配你期望的logs结构

4. $sort：

   • 最后按月份排序，让结果按时间顺序展示

虽然聚合管道是更优选择，但如果你一定要用MapReduce，这里也给你一个实现思路：
MapReduce需要先在map阶段提取月份和错误类型，reduce阶段收集同类型日志，最后还要在客户端手动整理成目标结构——步骤相对繁琐，性能也不如聚合管道，仅作参考：

rfidAttErrorSchema.mapReduce(
  // Map函数：输出「月份+错误类型」作为键，当前日志作为值
  function() {
    const month = new Date(this.error_date).getMonth() + 1; // 注意JS的getMonth是0-11，要+1转为1-12
    emit({ month: month, errorType: this.error_type }, this);
  },
  // Reduce函数：收集同组的所有日志文档
  function(key, values) {
    return values;
  },
  {
    query: condition,
    out: { inline: 1 } // 结果直接返回，不写入集合
  },
  function(err, results) {
    if (err) {
      return res.status(500).json({ result: err });
    }
    // 客户端手动按月份整理成目标结构
    const finalData = {};
    results.forEach(item => {
      const month = item._id.month;
      if (!finalData[month]) {
        finalData[month] = { _id: month, logs: {} };
      }
      finalData[month].logs[item._id.errorType] = item.value;
    });
    res.json({ result: Object.values(finalData) });
  }
);

内容的提问来源于stack exchange，提问作者kisor