MongoDB聚合管道：基于$lookup生成字段关联多集合统计数组匹配次数

你在聚合管道上一步生成的an_user_id字段可以直接被后续阶段调用，无需特殊处理，以下是完整实现方案：

完整聚合管道代码（pymongo适配） #

pipeline = [
    # 原有逻辑：关联members和an_users获取用户ID映射
    {
        '$lookup': {
            'from': 'an_users',                   
            'localField': 'membershipNumber',     
            'foreignField': 'memref',             
            'as': 'an_users'                      
        }
    },
    {'$unwind' : '$an_users'},     
    {   
        '$project' : {
            'membershipNumber': 1,
            'an_user_id' : '$an_users.user_id'
        } 
    },
    # 新增：关联emails集合统计邮件操作次数
    {
        '$lookup': {
            'from': 'emails',
            'let': {'current_uid': '$an_user_id'},
            'pipeline': [
                # 过滤当前用户有操作记录的邮件
                {
                    '$match': {
                        '$expr': {
                            '$or': [
                                {'$in': ['$$current_uid', '$click']},
                                {'$in': ['$$current_uid', '$bounce']},
                                {'$in': ['$$current_uid', '$opens']}
                            ]
                        }
                    }
                },
                # 分组统计各操作的总次数
                {
                    '$group': {
                        '_id': None,
                        'n_email_clicks': {
                            '$sum': {'$cond': [{'$in': ['$$current_uid', '$click']}, 1, 0]}
                        },
                        'n_email_bounces': {
                            '$sum': {'$cond': [{'$in': ['$$current_uid', '$bounce']}, 1, 0]}
                        },
                        'n_email_opens': {
                            '$sum': {'$cond': [{'$in': ['$$current_uid', '$opens']}, 1, 0]}
                        }
                    }
                }
            ],
            'as': 'email_stats'
        }
    },
    # 展开邮件统计结果，无匹配记录的用户保留
    {
        '$unwind': {
            'path': '$email_stats',
            'preserveNullAndEmptyArrays': True
        }
    },
    # 新增：关联events集合统计活动报名次数
    {
        '$lookup': {
            'from': 'events',
            'let': {'current_uid': '$an_user_id'},
            'pipeline': [
                # 过滤当前用户报名的活动
                {'$match': {'$expr': {'$in': ['$$current_uid', '$rsvps.rsvp_user_id']}}},
                # 统计报名总次数
                {'$count': 'n_rsvps'}
            ],
            'as': 'event_stats'
        }
    },
    # 展开活动统计结果，无匹配记录的用户保留
    {
        '$unwind': {
            'path': '$event_stats',
            'preserveNullAndEmptyArrays': True
        }
    },
    # 最终格式化输出字段，缺失统计值默认补0
    {
        '$project': {
            'membershipNumber': 1,
            'an_user_id': 1,
            'n_email_clicks': {'$ifNull': ['$email_stats.n_email_clicks', 0]},
            'n_email_bounces': {'$ifNull': ['$email_stats.n_email_bounces', 0]},
            'n_email_opens': {'$ifNull': ['$email_stats.n_email_opens', 0]},
            'n_rsvps': {'$ifNull': ['$event_stats.n_rsvps', 0]}
        }
    }
    # 若需要生成物理集合，取消下方注释，集合名可自行修改
    # , {'$out': 'user_member_statistics'}
]

# 执行聚合查询
result = members.aggregate(pipeline)

# 若需要生成实时查询视图（无物理存储），使用下方代码：
# db.create_view('user_member_stats_view', 'members', pipeline)

关键逻辑说明 #

• 管道执行规则：MongoDB聚合管道按阶段顺序执行，前序阶段输出的所有字段（包括计算生成的an_user_id）都可以直接作为后续操作的参数使用，无需额外适配。
• 子管道$lookup优势：通过let传参+子管道过滤统计的方式，避免全量拉取关联集合数据后再处理，大幅提升大数量级下的查询效率。
• 空值兼容：preserveNullAndEmptyArrays参数和$ifNull运算符配合，保证没有操作记录的用户不会被过滤，统计结果更完整。
• 结果持久化选择：
  • 数据量不大、需要实时查询的场景优先使用视图，无需占用额外存储空间。
  • 数据量大、需要离线分析的场景使用$out生成物理集合，查询性能更高。

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