MongoDB：如何用单个聚合管道实现多集合分组统计需求

需求：基于users集合，统计指定时间范围内每个用户在questions、answers、comments三个集合中的文档数量，结果需包含所有用户（即使某类文档数量为0）。

示例数据 #

db = {
    users: [{ _id: ObjectId('123') }, { _id: ObjectId('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') },
    ]
}

期望结果（时间范围：createdAt > ISODate('2022-09-03T00:00:00Z')） #

[
    { _id: ObjectId('123'), questionCount: 0, answerCount: 1, commentCount: 0 }, 
    { _id: ObjectId('456'), questionCount: 1, answerCount: 0, commentCount: 1 }
]

当前方案是对三个集合分别执行聚合后在后端合并，效率较低，以下是两种单次聚合的实现方式：

方案一：使用$lookup关联统计 #

从users集合出发，通过带管道的$lookup分别查询每个集合的符合条件的文档数量，最后统一处理结果：

db.users.aggregate([
    // 关联questions集合，统计符合时间条件的数量
    {
        $lookup: {
            from: "questions",
            let: { userId: "$_id" },
            pipeline: [
                { $match: {
                    $expr: { $eq: ["$authorId", "$$userId"] },
                    createdAt: { $gt: ISODate('2022-09-03T00:00:00Z') }
                }},
                { $count: "count" }
            ],
            as: "questionStats"
        }
    },
    // 提取questions的统计数，默认0
    {
        $addFields: {
            questionCount: {
                $ifNull: [{ $arrayElemAt: ["$questionStats.count", 0] }, 0]
            }
        }
    },
    // 关联answers集合，统计符合条件的数量
    {
        $lookup: {
            from: "answers",
            let: { userId: "$_id" },
            pipeline: [
                { $match: {
                    $expr: { $eq: ["$authorId", "$$userId"] },
                    createdAt: { $gt: ISODate('2022-09-03T00:00:00Z') }
                }},
                { $count: "count" }
            ],
            as: "answerStats"
        }
    },
    // 提取answers的统计数，默认0
    {
        $addFields: {
            answerCount: {
                $ifNull: [{ $arrayElemAt: ["$answerStats.count", 0] }, 0]
            }
        }
    },
    // 关联comments集合，统计符合条件的数量
    {
        $lookup: {
            from: "comments",
            let: { userId: "$_id" },
            pipeline: [
                { $match: {
                    $expr: { $eq: ["$authorId", "$$userId"] },
                    createdAt: { $gt: ISODate('2022-09-03T00:00:00Z') }
                }},
                { $count: "count" }
            ],
            as: "commentStats"
        }
    },
    // 提取comments的统计数，默认0
    {
        $addFields: {
            commentCount: {
                $ifNull: [{ $arrayElemAt: ["$commentStats.count", 0] }, 0]
            }
        }
    },
    // 保留需要的字段
    {
        $project: {
            questionCount: 1,
            answerCount: 1,
            commentCount: 1
        }
    }
])

方案二：使用$unionWith合并后统计 #

先合并三个集合的数据并标记类型，统计后再关联users补全所有用户：

db.users.aggregate([
    // 合并三个集合的数据，添加类型标记
    {
        $unionWith: {
            coll: "questions",
            pipeline: [
                { $match: { createdAt: { $gt: ISODate('2022-09-03T00:00:00Z') } } },
                { $addFields: { type: "question" } }
            ]
        }
    },
    {
        $unionWith: {
            coll: "answers",
            pipeline: [
                { $match: { createdAt: { $gt: ISODate('2022-09-03T00:00:00Z') } } },
                { $addFields: { type: "answer" } }
            ]
        }
    },
    {
        $unionWith: {
            coll: "comments",
            pipeline: [
                { $match: { createdAt: { $gt: ISODate('2022-09-03T00:00:00Z') } } },
                { $addFields: { type: "comment" } }
            ]
        }
    },
    // 按用户和类型分组统计数量
    {
        $group: {
            _id: { userId: "$authorId", type: "$type" },
            count: { $sum: 1 }
        }
    },
    // 按用户合并各类型统计结果
    {
        $group: {
            _id: "$_id.userId",
            stats: {
                $push: { k: "$_id.type", v: "$count" }
            }
        }
    },
    // 将stats数组转为键值对
    {
        $replaceRoot: {
            newRoot: {
                $mergeObjects: [
                    { _id: "$_id" },
                    { $arrayToObject: "$stats" }
                ]
            }
        }
    },
    // 关联users集合，确保所有用户都在结果中
    {
        $lookup: {
            from: "users",
            localField: "_id",
            foreignField: "_id",
            as: "user"
        }
    },
    { $unwind: "$user" },
    // 填充缺失的类型计数为0
    {
        $addFields: {
            questionCount: { $ifNull: ["$question", 0] },
            answerCount: { $ifNull: ["$answer", 0] },
            commentCount: { $ifNull: ["$comment", 0] }
        }
    },
    // 保留目标字段
    {
        $project: {
            _id: "$user._id",
            questionCount: 1,
            answerCount: 1,
            commentCount: 1
        }
    }
])

两种方案中，方案一逻辑更直观，适合集合数据量不大的场景；方案二更适合需要批量处理多集合数据的场景，可根据实际数据规模选择。

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