MongoDB 4.4能否预存聚合管道并传入变量执行？

问题背景 #

我有一个基于PyMongo 3.10.1的Python应用，在MongoDB 4.4服务器上运行包含约18个阶段的聚合管道，最终阶段为$merge（无结果返回客户端）。尽管管道规模较大，但仅访问少量文档，执行速度极快。

我通过32个线程调用该管道约一百万次，每次调用仅需修改第一个$match阶段的参数值：

pipeline = [
    {
        '$match': {
            'week': {'$in': <var-list1>},
            'eventCode': <var2>,
            'customer': <var3>
        }
    }, 
    # 固定的一系列聚合阶段
    {
        '$merge': {
            'into': 'my_out_collection',
            'on': ['week', 'eventCode', 'customer'],
            'whenMatched': 'replace',
            'whenNotMatched': 'insert'
        }
    }
]

当前MongoDB服务器的磁盘、CPU仍有富余，但客户端网络出站流量已达上限，怀疑是每次调用都传输整个大管道导致的。想知道MongoDB 4.4是否支持预存带占位符的管道，仅传变量执行？

可行方案 #

1. 利用视图简化管道传输（推荐，无需启用JS） #

将除$match外的固定聚合阶段创建为视图，后续客户端仅需发送简短的$match阶段即可：

步骤1：创建视图（MongoDB Shell执行）

db.createView(
    "my_agg_view",
    "your_source_collection", // 替换为你的源集合名
    [
        // 这里放入所有固定的聚合阶段（去掉开头的$match）
        // 示例：假设固定阶段包含$group、$project等
        // { '$group': { ... } },
        // { '$project': { ... } },
        {
            '$merge': {
                'into': 'my_out_collection',
                'on': ['week', 'eventCode', 'customer'],
                'whenMatched': 'replace',
                'whenNotMatched': 'insert'
            }
        }
    ]
)

步骤2：PyMongo中调用视图

每次仅需传递$match参数，管道体积大幅缩小：

from pymongo import MongoClient

client = MongoClient()
db = client.your_database

# 每次调用仅构造简短的$match管道
match_pipeline = [
    {
        '$match': {
            'week': {'$in': week_list_var},
            'eventCode': event_code_var,
            'customer': customer_var
        }
    }
]

# 对视图执行聚合，无需传输完整大管道
db.my_agg_view.aggregate(match_pipeline)

2. 服务器端存储JavaScript函数（兼容4.4） #

将完整管道定义在服务器端的JS函数中，客户端仅传递变量调用：

步骤1：存储函数（MongoDB Shell执行）

db.system.js.save({
    _id: "runMyAggregation",
    value: function(weekList, eventCodeVal, customerVal) {
        const pipeline = [
            {
                '$match': {
                    'week': {'$in': weekList},
                    'eventCode': eventCodeVal,
                    'customer': customerVal
                }
            },
            // 放入所有固定聚合阶段
            {
                '$merge': {
                    'into': 'my_out_collection',
                    'on': ['week', 'eventCode', 'customer'],
                    'whenMatched': 'replace',
                    'whenNotMatched': 'insert'
                }
            }
        ];
        return db.your_source_collection.aggregate(pipeline);
    }
})

步骤2：PyMongo中调用函数

from pymongo import MongoClient

client = MongoClient()
db = client.your_database

# 仅传递变量，无需传输管道
db.command('eval', 'runMyAggregation(?, ?, ?)', args=[week_list_var, event_code_var, customer_var])

注意：eval命令在MongoDB后续版本中已被标记为废弃，但4.4版本可正常使用；需确保服务器启用了JavaScript执行（默认开启）。

额外优化建议 #

• 批量合并请求：将多个$match参数打包成批量任务，减少网络调用次数。
• 建立复合索引：为week、eventCode、customer创建复合索引，提升$match阶段的查询效率。

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