MongoDB：如何从具有相同聚合管道的多个集合中聚合答案

【问题标题】：MongoDB: how to aggregate from multiple collections with same aggregation pipelineMongoDB：如何从具有相同聚合管道的多个集合中聚合
【发布时间】：2022-09-23 16:22:36
【问题描述】：

我正在尝试使用相同的聚合管道获取聚合，包括来自多个集合的 $match 和 $group 操作。

例如，

带有users 集合和questions、answers 和comments 的集合，其中每个文档都有authorId 和created_at 字段，

db = [
    \'users\': [{ _id: 123 }, { _id: 456} ],
    \'questions\': [
        { authorId: ObjectId(\'123\'), createdAt: ISODate(\'2022-09-01T00:00:00Z\') },
        { authorId: ObjectId(\'456\'), createdAt: ISODate(\'2022-09-05T00:00:00Z\') },
    ],

    \'answers\': [
        { authorId: ObjectId(\'123\'), createdAt: ISODate(\'2022-09-05T08:00:00Z\') },
        { authorId: ObjectId(\'456\'), createdAt: ISODate(\'2022-09-01T08:00:00Z\') },
    ],
    \'comments\': [
        { authorId: ObjectId(\'123\'), createdAt: ISODate(\'2022-09-01T16:00:00Z\') },
        { authorId: ObjectId(\'456\'), createdAt: ISODate(\'2022-09-05T16:00:00Z\') },
    ],
]

我想从每个集合中获取文档计数，其中created_at 在给定范围之间并按authorId 分组。所需的聚合结果可能如下所示。这里的_ids 是users 集合中文档的ObjectId。

\\\\ match: { createdAt: { $gt: ISODate(\'2022-09-03T00:00:00Z) } }
[
    { _id: ObjectId(\'123\'), questionCount: 0, answerCount: 1, commentCount: 0 }, 
    { _id: ObjectId(\'456\'), questionCount: 1, answerCount: 0, commentCount: 1 }
]

目前，我在下面为每个集合运行聚合，结合后端服务中的结果。（我正在使用 Spring Data MongoDB Reactive。）这似乎非常低效。

db.collection.aggregate([
    { $match: { 
        created_at: { $gt: ISODate(\'2022-09-03T00:00:00Z\') }
    }},
    { $group : {
        _id: \'$authorId\',
        count: {$sum: 1}
    }}
])

如何通过一次聚合获得所需的结果？

我认为$unionWith 或$lookup 可能会有所帮助，但我被困在这里。

标签： mongodb mongodb-query aggregation-framework spring-data-mongodb spring-data-mongodb-reactive

【解决方案1】：

您可以尝试这样的事情，使用 $lookup，在这里我们加入用户，所有三个集合一个接一个，然后计算计数：

db.users.aggregate([
  {
    "$lookup": {
      "from": "questions",
      "let": {
        id: "$_id"
      },
      "pipeline": [
        {
          "$match": {
            $expr: {
              "$and": [
                {
                  "$gt": [
                    "$createdAt",
                    ISODate("2022-09-03T00:00:00Z")
                  ]
                },
                {
                  "$eq": [
                    "$$id",
                    "$authorId"
                  ]
                }
              ]
            }
          }
        }
      ],
      "as": "questions"
    }
  },
  {
    "$lookup": {
      "from": "answers",
      "let": {
        id: "$_id"
      },
      "pipeline": [
        {
          "$match": {
            $expr: {
              "$and": [
                {
                  "$gt": [
                    "$createdAt",
                    ISODate("2022-09-03T00:00:00Z")
                  ]
                },
                {
                  "$eq": [
                    "$$id",
                    "$authorId"
                  ]
                }
              ]
            }
          }
        }
      ],
      "as": "answers"
    }
  },
  {
    "$lookup": {
      "from": "comments",
      "let": {
        id: "$_id"
      },
      "pipeline": [
        {
          "$match": {
            $expr: {
              "$and": [
                {
                  "$gt": [
                    "$createdAt",
                    ISODate("2022-09-03T00:00:00Z")
                  ]
                },
                {
                  "$eq": [
                    "$$id",
                    "$authorId"
                  ]
                }
              ]
            }
          }
        }
      ],
      "as": "comments"
    }
  },
  {
    "$project": {
      "questionCount": {
        "$size": "$questions"
      },
      "answersCount": {
        "$size": "$answers"
      },
      "commentsCount": {
        "$size": "$comments"
      }
    }
  }
])

游乐场link。在上面的查询中，我们使用 $lookup 的流水线形式，对一些自定义逻辑执行连接。在此处了解有关$lookup 的更多信息。

另一种方法是，执行正常查找，然后过滤掉元素：

db.users.aggregate([
  {
    "$lookup": {
      "from": "questions",
      "localField": "_id",
      "foreignField": "authorId",
      "as": "questions"
    }
  },
  {
    "$lookup": {
      "from": "answers",
      "localField": "_id",
      "foreignField": "authorId",
      "as": "answers"
    }
  },
  {
    "$lookup": {
      "from": "comments",
      "localField": "_id",
      "foreignField": "authorId",
      "as": "comments"
    }
  },
  {
    "$project": {
      questionCount: {
        "$size": {
          "$filter": {
            "input": "$questions",
            "as": "item",
            "cond": {
              "$gt": [
                "$$item.createdAt",
                ISODate("2022-09-03T00:00:00Z")
              ]
            }
          }
        }
      },
      answerCount: {
        "$size": {
          "$filter": {
            "input": "$answers",
            "as": "item",
            "cond": {
              "$gt": [
                "$$item.createdAt",
                ISODate("2022-09-03T00:00:00Z")
              ]
            }
          }
        }
      },
      commentsCount: {
        "$size": {
          "$filter": {
            "input": "$comments",
            "as": "item",
            "cond": {
              "$gt": [
                "$$item.createdAt",
                ISODate("2022-09-03T00:00:00Z")
              ]
            }
          }
        }
      }
    }
  }
])

游乐场link。

【讨论】：

感谢您的回答，但是我应该将$match（或created_at 的任何过滤操作）放在管道中的哪个位置？
它应该是第一阶段，在$lookup 之前。也更新了答案。
我可以在使用$lookup 加入文档之前添加过滤器吗？正如我在问题中提到的，我正在尝试过滤除users 之外的其他集合中的文档。
好的，只能在 $lookup 内完成。您可以将所有集合中的一些示例文档添加到问题本身吗？我将相应地更新查询。
酷，我会尽快更新答案