基于参考映射计算子文档列表的最大类别值（MongoDB聚合）

原始MongoDB文档结构 #

[
    {
        "country": "UK",
        "shops": [
            {"city": "London", "fruits": ["banana", "apple"]},
            {"city": "Birmingham", "fruits": ["banana", "pineapple"]}
        ]
    },
    {
        "country": "DE",
        "shops": [
            {"city": "Munich", "fruits": ["banana", "strawberry"]},
            {"city": "Berlin", "fruits": ["kiwi", "pineapple"]}
        ]
    }
]

水果-类别映射字典（Python中定义） #

categories = {
    1: ["apple"],
    2: ["banana", "kiwi"],
    3: ["pineapple", "strawberry"]
}

期望输出 #

[
    {
        "country": "UK",
        "shops": [
            {"city": "London", "fruits": ["banana", "apple"]},
            {"city": "Birmingham", "fruits": ["banana", "pineapple"]}
        ],
        "max_category": 3
    },
    {
        "country": "DE",
        "shops": [
            {"city": "Munich", "fruits": ["banana", "apple"]},
            {"city": "Berlin", "fruits": ["kiwi", "apple"]}
        ],
        "max_category": 2
    }
]

聚合管道实现方案 #

核心思路是先扁平化嵌套的水果列表，匹配每个水果对应的类别，最后聚合计算类别最大值。具体步骤如下：

完整聚合管道 #

db.collection.aggregate([
    // 展开shops数组，将每个店铺拆分为独立文档
    { $unwind: "$shops" },
    // 展开每个店铺的fruits数组，将每个水果拆分为独立文档
    { $unwind: "$shops.fruits" },
    // 为每个水果匹配对应类别ID
    {
        $project: {
            country: 1,
            shops: 1,
            category: {
                $switch: {
                    branches: [
                        { case: { $in: ["$shops.fruits", ["apple"]] }, then: 1 },
                        { case: { $in: ["$shops.fruits", ["banana", "kiwi"]] }, then: 2 },
                        { case: { $in: ["$shops.fruits", ["pineapple", "strawberry"]] }, then: 3 }
                    ],
                    default: 0
                }
            }
        }
    },
    // 按原文档ID聚合，收集所有类别并计算最大值，同时还原原始字段
    {
        $group: {
            _id: "$_id",
            country: { $first: "$country" },
            shops: { $push: "$shops" },
            max_category: { $max: "$category" }
        }
    },
    // 去重重复的shop条目（展开后会生成重复项）
    {
        $addFields: {
            unique_shops: { $addToSet: "$shops" }
        }
    },
    // 整理输出字段，替换为去重后的shops数组
    {
        $project: {
            _id: 1,
            country: 1,
            shops: "$unique_shops",
            max_category: 1
        }
    }
])

Python中动态生成管道的代码 #

为避免硬编码类别匹配规则，可通过Python字典动态生成聚合管道的分支逻辑：

from pymongo import MongoClient

# 连接MongoDB
client = MongoClient("mongodb://localhost:27017/")
db = client["your_database_name"]
collection = db["your_collection_name"]

# 定义水果类别映射
categories = {
    1: ["apple"],
    2: ["banana", "kiwi"],
    3: ["pineapple", "strawberry"]
}

# 动态生成$switch的匹配分支
switch_branches = []
for cat_id, fruit_list in categories.items():
    switch_branches.append({
        "case": {"$in": ["$shops.fruits", fruit_list]},
        "then": cat_id
    })

# 构建聚合管道
pipeline = [
    {"$unwind": "$shops"},
    {"$unwind": "$shops.fruits"},
    {
        "$project": {
            "country": 1,
            "shops": 1,
            "category": {
                "$switch": {
                    "branches": switch_branches,
                    "default": 0
                }
            }
        }
    },
    {
        "$group": {
            "_id": "$_id",
            "country": {"$first": "$country"},
            "shops": {"$push": "$shops"},
            "max_category": {"$max": "$category"}
        }
    },
    {
        "$addFields": {
            "unique_shops": {"$addToSet": "$shops"}
        }
    },
    {
        "$project": {
            "_id": 1,
            "country": 1,
            "shops": "$unique_shops",
            "max_category": 1
        }
    }
]

# 执行聚合并打印结果
result = list(collection.aggregate(pipeline))
for doc in result:
    print(doc)

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