MongoDB聚合管道优化：6000+文档查询超时问题求助

问题背景 #

使用M10 Atlas集群，执行以下聚合管道处理约6000条文档时耗时超3分钟，导致请求超时：

[
  {
    $match: {
      "cohortId": payload.cohortId, "template": req.query.template
    },
  }, {
    $lookup: {
      from: "messages", localField: "messageId", foreignField: "metaData.messageId", as: "message"
    }
  }, {
    $lookup: {
      from: "webhook", localField: "messageId", foreignField: "messageId", as: "webhook"
    }
  }, {
    $unwind: {
      path: "$webhook"
    }
  }, {
    $unwind: {
      path: "$message"
    }
  },
  {
    $project: {
      "_id": 0,
      "messageId": 1,
      "cohortId": 1,
      "template": 1,
      "origin": 1,
      "WBA_AccountId": 1,
      "WBA_PhoneId": 1,
      "clientPhone": "$phone",
      "messageDirection": "$message.metaData.direction",
      "messageTime": "$message.metaData.time",
      "messageSent": "$webhook.status.sentFlag",
      "messageDelivered": "$webhook.status.deliveredFlag",
      "messageRead": "$webhook.status.readFlag",
      "messageFailed": "$webhook.status.failedFlag",
      "messageFailedReason": "$webhook.status.failedReason",
      "messageSentTime": "$webhook.status.sentTimestamp",
      "messageDeliveredTime": "$webhook.status.deliveredTimestamp",
      "messageReadTime": "$webhook.status.readTimestamp",
      "messageFailedTime": "$webhook.status.failedTimestamp"
    }
  }
]

涉及集合结构：

• webhook集合：

{
  "_id": ObjectId("6374d59618bff45fa34f08f5"),
  "messageId": "<Message ID as String>",
  "conversationExpiry": 1668687660,
  "conversationId": "<Conversation ID as String>",
  "billableFlag": "true",
  "WBA_PhoneId": 1234567890,
  "WBA_AccountId": 1234567890,
  "WBA_DisplayPhone": 1234567890,
  "phone": 1234567890,
  "status": {
    "sentFlag": true,
    "sentTimestamp": 1668601237,
    "deliveredFlag": true,
    "readFlag": true,
    "failedFlag": false,
    "deliveredTimestamp": 1668601238,
    "readTimestamp": 1668601250,
    "failedTimestamp": 0,
    "errorMessage": "",
    "errorCode": ""
  },
  "createdAt": ISODate("2022-11-16T12:20:38.086Z"),
  "updatedAt": ISODate("2022-11-16T12:20:50.918Z")
}

• audience集合：

{
  "_id": ObjectId("635a840b97405992d3cb794d"),
  "WBA_AccountId": 1234567890,
  "WBA_PhoneId": 1234567890,
  "messageId": "<Message ID as String>",
  "phone": 1234567890,
  "cohortId": "<String Value>",
  "createdAt": ISODate("2022-10-27T12:33:47.333Z"),
  "end": "2022-10-27",
  "origin": "Clevertap_API_Campaigns",
  "start": "2022-10-27",
  "template": "<String Value>",
  "updatedAt": ISODate("2022-10-27T12:33:47.333Z")
}

优化方案 #

1. 创建针对性索引（核心优化） #

索引是提升聚合性能的关键，针对过滤和关联字段创建索引：

• audience集合：创建复合索引覆盖$match过滤和后续$lookup的关联字段，减少文档扫描范围：

  db.audience.createIndex({ cohortId: 1, template: 1, messageId: 1 })
  

• messages集合：针对$lookup的关联字段创建单字段索引：

  db.messages.createIndex({ "metaData.messageId": 1 })
  

• webhook集合：针对$lookup的关联字段创建单字段索引：

  db.webhook.createIndex({ messageId: 1 })
  

2. 优化$lookup阶段，减少数据传输 #

当前$lookup会拉取整个关联文档，可通过内部管道仅投影需要的字段，大幅降低数据量：

• 修改messages的$lookup：

  {
    $lookup: {
      from: "messages",
      localField: "messageId",
      foreignField: "metaData.messageId",
      as: "message",
      pipeline: [
        { $project: { "metaData.direction": 1, "metaData.time": 1, _id: 0 } }
      ]
    }
  }
  

• 修改webhook的$lookup：

  {
    $lookup: {
      from: "webhook",
      localField: "messageId",
      foreignField: "messageId",
      as: "webhook",
      pipeline: [
        { $project: { 
          "status.sentFlag": 1, 
          "status.deliveredFlag": 1, 
          "status.readFlag": 1, 
          "status.failedFlag": 1, 
          "status.failedReason": 1, 
          "status.sentTimestamp": 1, 
          "status.deliveredTimestamp": 1, 
          "status.readTimestamp": 1, 
          "status.failedTimestamp": 1, 
          _id: 0 
        } }
      ]
    }
  }
  

3. 提前投影缩减数据体积 #

在$match之后立即添加轻量投影，仅保留后续阶段需要的字段，减少数据处理量：

{
  $project: {
    _id: 0,
    messageId: 1,
    cohortId: 1,
    template: 1,
    origin: 1,
    WBA_AccountId: 1,
    WBA_PhoneId: 1,
    phone: 1
  }
}

将此阶段插入到$match之后、第一个$lookup之前。

4. 检查Atlas集群资源与查询分析 #

• 查看集群CPU、内存负载：若M10集群资源饱和，可临时升级到M20或更高规格，或排查是否有其他高负载查询抢占资源。
• 使用Atlas查询分析器：定位管道中耗时最长的阶段，针对性优化。

5. 异步处理或分页返回（可选） #

若业务允许：

• 采用异步预计算：定时执行聚合任务，将结果存储到汇总集合（如audience_message_summary），控制器直接查询汇总集合返回数据。
• 分页返回：通过$skip和$limit拆分结果，减少单次请求的处理量，避免超时。

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