跨MongoDB双集合关联查询的Python实现异常排查求助

我有两个MongoDB集合：LockQueue和MessageClassification，其中MessageClassification的_id以UUID字符串形式存储在LockQueue的Data.ClassificationId字段中。需求是通过MessageClassification集合里Message.EventId等于ESI00008的条件作为入口，获取LockQueue集合中的对应数据。

在Studio 3T中执行以下查询可得到预期结果：

db.getCollection("MessageClassification").find({'Message.EventId': 'ESI00008'})
db.getCollection("LockQueue").find({"Data.ClassificationId": CSUUID("EDDF6494-D4BB-48C2-A08E-D61F31313826")})

用Python实现自动化查询时，写出了如下代码，但运行时无报错却无限运行或速度极慢，甚至Ctrl+C都无法终止（连接字符串与Studio 3T使用的一致）：

from pymongo import MongoClient
import uuid

uri = "mongodb://user:password@server.intern:27017/DB?serverSelectionTimeoutMS=5000&connectTimeoutMS=10000&authSource=admin&authMechanism=SCRAM-SHA-1"

client = MongoClient(uri, uuidRepresentation='csharpLegacy')


mydatabase = client["DB"]

coll = mydatabase["LockQueue"]

cursor = [{'$lookup': {
             'from': 'MessageClassification',
             'let': {'id': {'$toString': '$_id'}},
             'pipeline': [
               {
                 '$match': {
                   '$expr': {
                     '$and': [
                       {'$eq': ['$$id', '$Data.ClassificationId']},
                       {'$eq': ['$Message.EventId', 'ESI00008']}
                     ]                       
                    }
                  }
                }
             ],
             'as': 'message'
           }
         }]
           
for doc in (coll.aggregate(cursor)):
    print(doc)

核心问题分析 #

1. 聚合方向错误：原代码从LockQueue集合发起聚合查询，会遍历集合中所有文档，逐个与MessageClassification关联。如果LockQueue数据量很大，这种方式会导致性能急剧下降，甚至出现无限运行的情况。正确逻辑应该是先从MessageClassification筛选出符合EventId的小批量数据，再关联LockQueue。
2. 匹配逻辑颠倒：原$lookup中的匹配条件写反了——MessageClassification的_id应该对应LockQueue的Data.ClassificationId，但原代码错误地将LockQueue的_id转字符串后与MessageClassification的Data.ClassificationId对比，完全不符合业务逻辑。
3. 缺少索引（潜在问题）：如果MessageClassification.Message.EventId和LockQueue.Data.ClassificationId没有建立索引，全表扫描会进一步加剧性能问题。

修复后的代码 #

方式一：分步查询（直观高效，适合数据量不大的场景） #

先查询MessageClassification得到符合条件的_id，再用这些_id去查询LockQueue：

from pymongo import MongoClient
import uuid

uri = "mongodb://user:password@server.intern:27017/DB?serverSelectionTimeoutMS=5000&connectTimeoutMS=10000&authSource=admin&authMechanism=SCRAM-SHA-1"

client = MongoClient(uri, uuidRepresentation='csharpLegacy')
db = client["DB"]

# 第一步：获取符合EventId的MessageClassification文档的_id
mc_coll = db["MessageClassification"]
target_ids = [doc["_id"] for doc in mc_coll.find({"Message.EventId": "ESI00008"}, {"_id": 1})]

# 第二步：用获取到的_id查询LockQueue
lq_coll = db["LockQueue"]
for doc in lq_coll.find({"Data.ClassificationId": {"$in": target_ids}}):
    print(doc)

方式二：正确的聚合查询（适合复杂关联场景） #

调整聚合方向，从MessageClassification发起查询，关联LockQueue：

from pymongo import MongoClient
import uuid

uri = "mongodb://user:password@server.intern:27017/DB?serverSelectionTimeoutMS=5000&connectTimeoutMS=10000&authSource=admin&authMechanism=SCRAM-SHA-1"

client = MongoClient(uri, uuidRepresentation='csharpLegacy')
db = client["DB"]

mc_coll = db["MessageClassification"]

cursor = mc_coll.aggregate([
    # 先筛选出符合EventId的文档
    {"$match": {"Message.EventId": "ESI00008"}},
    # 关联LockQueue，用当前文档的_id匹配LockQueue的Data.ClassificationId
    {"$lookup": {
        "from": "LockQueue",
        "localField": "_id",
        "foreignField": "Data.ClassificationId",
        "as": "lock_queue_docs"
    }}
])

for doc in cursor:
    # 输出关联到的LockQueue数据
    for lq_doc in doc["lock_queue_docs"]:
        print(lq_doc)

性能优化建议 #

为以下字段创建索引，大幅提升查询速度：

// 给MessageClassification的Message.EventId创建索引
db.MessageClassification.createIndex({"Message.EventId": 1})

// 给LockQueue的Data.ClassificationId创建索引
db.LockQueue.createIndex({"Data.ClassificationId": 1})

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