MongoDB百万级数据批量迁移优化咨询（Django+AWS环境）

我的MongoDB leads集合有约190万条记录，样本文档结构如下：

{
  "_id": {
    "$oid": "60174f6cc3"
  },
  "customer_id": "590583",
  "email": "xyz@gmail.com",
  "firstname": "XYZ",
  "mobile": "234567876",
  "newsletter": "1",
  "verified_email": "0",
  "verified_mobile": "1",
  "luid": "0zmapjn1-ij3h-aq-jifg-x0w3lvkkm8mbzx",
  "createdAt": {
    "$date": "2021-02-01T07:46:08.000Z"
  },
  "modifiedAt": {
    "$date": "2022-08-01T09:39:15.000Z"
  },
  "identifiedAt": {
    "$date": "2021-02-01T08:01:32.201Z"
  },
  "segments": [
    "64ec38d66794a663c71fb7c1"
  ]
}

基于用户行为创建分群后，需将匹配指定segment ID的文档写入campaignAudience集合。小受众规模时运行正常，但受众达80万、120万以上时，数据获取与存储耗时过长或超时。已尝试线程处理存储，但数据获取仍有问题；考虑分批处理但担心效率低下。当前使用Django 3.2，MongoDB部署在AWS EC2，求最优实现方案，是否需要Spark、Hadoop等技术？

1. 基础优化：给segments字段建多键索引 #

这是提升查询速度的核心前提，无索引时查询会触发全表扫描，190万条数据的扫描极慢。执行以下MongoDB命令：

db.leads.createIndex({ segments: 1 })

索引创建完成后，匹配segment ID的查询速度会提升几个数量级。

2. 优先用MongoDB原生聚合完成数据复制（最快方案） #

无需通过应用层中转数据，直接让MongoDB在数据库内部完成筛选和写入，避免数据传输开销，这是效率最高的方式：

• 若需直接覆盖campaignAudience集合（适合每次创建新营销活动时清空目标集合的场景）：

db.leads.aggregate([
  { $match: { segments: "目标segment ID" } },
  { $out: "campaignAudience" }
])

• 若需增量写入（比如给已有受众添加新用户）：

db.leads.aggregate([
  { $match: { segments: "目标segment ID" } },
  { $merge: { into: "campaignAudience" } }
])

可以通过Django调用pymongo执行这段聚合逻辑，无需复杂的应用层代码。

3. 需应用层处理时：游标分批查询+原生批量写入 #

如果需要在写入前做业务逻辑处理（比如字段转换、额外校验），则用MongoDB游标分批拉取数据，避免一次性加载所有数据到内存，再用批量写入接口提交。示例代码（直接用pymongo，比Django ORM高效）：

from pymongo import MongoClient, ReadPreference

# 建立连接，可指定读偏好到从节点，减轻主节点压力
client = MongoClient("你的MongoDB连接字符串")
db = client["你的数据库名"]

# 用游标查询，batch_size控制每次从数据库拉取的条数
cursor = db.leads.find(
    {"segments": "目标segment ID"},
    batch_size=10000,
    read_preference=ReadPreference.SECONDARY  # 有副本集时启用
)

batch = []
batch_size = 10000
total_written = 0

for doc in cursor:
    # 在这里加入你的业务处理逻辑，比如字段修改、过滤等
    # doc["extra_field"] = "some_value"
    
    batch.append(doc)
    if len(batch) >= batch_size:
        db.campaignAudience.insert_many(batch)
        total_written += len(batch)
        batch = []
        print(f"已写入{total_written}条")

# 处理剩余的最后一批数据
if batch:
    db.campaignAudience.insert_many(batch)
    total_written += len(batch)
    print(f"写入完成，共{total_written}条")

4. 异步化处理，避免请求超时 #

将数据同步任务放到后台执行，不要在Django的同步请求中处理，避免HTTP超时。可以用Celery作为任务队列，把上述批量处理逻辑封装成Celery任务，通过Django视图触发任务执行，前端轮询或接收回调获取进度。

5. 优化MongoDB与EC2配置 #

• 调整MongoDB的WiredTiger缓存大小，比如EC2内存为8GB时，设置wiredTigerCacheSizeGB=4（在mongod.conf中配置），让MongoDB有足够缓存存储热点数据。
• 确保EC2实例的CPU、内存配置匹配数据规模，190万条数据建议至少用t3.medium及以上实例，避免资源瓶颈。
• 如果是副本集部署，将查询请求分流到从节点，减轻主节点的读写压力。

不需要。190万条数据属于中等规模，MongoDB原生的聚合、批量操作完全可以高效处理，引入Spark/Hadoop会增加系统复杂度和运维成本，属于过度设计。只有当数据规模达到TB级以上，或者需要复杂的分布式计算时，才考虑这类大数据框架。

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