MongoDB嵌套数组多分组并按位置限制街道数量的高效聚合方案

原始文档结构 #

{
  "id": "1",
  "category": "education",
  "field": "science",
  "items": [
    {"location": "mumbai","street": "Marine Drive"},
    {"location": "mumbai","street": "Linking Road"},
    {"location": "london","street": "Oxford Street"},
    {"location": "mumbai","street": "Bruce Street"},
    {"location": "delhi","street": "Chandni Chowk"},
    {"location": "london","street": "Carnaby Street"},
    {"location": "london","street": "King's Road"}
  ]
}

当前使用的聚合管道 #

aggregate([
  { $unwind: "$items" },
  { $group: { _id: { location: "$items.location" }, street: { $push: "$items.street" } } },
  { $project: { _id: 0, location: "$_id.location", street: { $filter: { input: "$street", as: "street", cond: { $lt: [ { $indexOfArray: [ "$street", "$$street" ] }, 2 ] } } } }
])

当前方案依赖$unwind展开数组再分组，当items数组元素较多时，会生成大量中间文档，性能损耗明显。

优化后的高效聚合管道 #

无需$unwind，直接在原数组上完成分组和条数限制，性能更优：

aggregate([
  {
    $project: {
      items: {
        $reduce: {
          input: "$items",
          initialValue: {},
          in: {
            $mergeObjects: [
              "$$value",
              {
                "$$this.location": {
                  $cond: [
                    { $lt: [ { $size: { $ifNull: [ "$$value.$$this.location", [] ] } }, 2 ] },
                    { $concatArrays: [ { $ifNull: [ "$$value.$$this.location", [] ] }, [ "$$this.street" ] ] },
                    "$$value.$$this.location"
                  ]
                }
              }
            ]
          }
        }
      }
    }
  },
  {
    $project: {
      items: {
        $map: {
          input: { $objectToArray: "$items" },
          as: "entry",
          in: { location: "$$entry.k", street: "$$entry.v" }
        }
      }
    }
  }
])

管道逻辑说明 #

1. 第一阶段$project：通过$reduce遍历items数组，以location为键构建对象。每个键对应的街道数组长度小于2时，才将当前街道加入数组；达到2条后不再新增。
2. 第二阶段$project：将上一步生成的键值对对象，通过$objectToArray转换为数组，再用$map映射成预期的{location, street}格式。

Node.js环境中的使用示例 #

// 假设已获取MongoDB集合实例
const db = await MongoClient.connect('your-mongodb-uri');
const collection = db.collection('your-collection-name');

const result = await collection.aggregate([
  {
    $project: {
      items: {
        $reduce: {
          input: "$items",
          initialValue: {},
          in: {
            $mergeObjects: [
              "$$value",
              {
                "$$this.location": {
                  $cond: [
                    { $lt: [ { $size: { $ifNull: [ "$$value.$$this.location", [] ] } }, 2 ] },
                    { $concatArrays: [ { $ifNull: [ "$$value.$$this.location", [] ] }, [ "$$this.street" ] ] },
                    "$$value.$$this.location"
                  ]
                }
              }
            ]
          }
        }
      }
    }
  },
  {
    $project: {
      items: {
        $map: {
          input: { $objectToArray: "$items" },
          as: "entry",
          in: { location: "$$entry.k", street: "$$entry.v" }
        }
      }
    }
  }
]).toArray();

console.log(result[0].items);

执行后会输出符合预期的结构：

[
  {"location": "mumbai","street": ["Marine Drive","Linking Road"]},
  {"location": "london","street": ["Oxford Street","Carnaby Street"]},
  {"location": "delhi","street": ["Chandni Chowk"]}
]

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