使用pymongo读取MongoDB数据生成DataFrame时的性能问题求助

处理500万条、4.7GB规模的文档确实容易遇到性能瓶颈，尤其是直接转Pandas DataFrame的时候。结合你给出的客户分层文档结构，我整理了几个实操性拉满的优化方向，都是我之前处理类似量级数据时亲测有效的方法：

一、先从MongoDB查询环节砍耗时 #

数据读取的源头慢，后面再怎么优化都是白搭，先把这一步搞定：

• 只拉取需要的字段：你的文档有多层CUST_LEVEL字段，如果不需要全部，一定要用投影（projection）过滤掉无关字段，比如只保留用到的层级，同时去掉默认返回的_id：

  # 只返回需要的字段，_id设为0表示不返回
  cursor = coll.find({}, {"CUST_LEVEL1":1, "CUST_LEVEL2":1, "CUST_LEVEL3":1, "CUST_LEVEL4":1, "CUST_LEVEL5":1, "_id":0})
  

  这一步能直接减少传输的数据量，4.7GB的全量数据砍到几分之一都有可能。
• 给查询字段加索引：如果你的查询有过滤条件（比如按CUST_LEVEL1筛选特定渠道），立刻给对应字段建索引：

  # 在Mongo shell或者PyMongo里执行都可以
  coll.create_index({"CUST_LEVEL1": 1})
  

  要是没有过滤条件，也可以给需要投影的字段建覆盖索引，这样Mongo不用回表读取完整文档，直接从索引里拿数据，速度能提升好几倍：

  coll.create_index({"CUST_LEVEL1":1, "CUST_LEVEL2":1, "CUST_LEVEL3":1, "CUST_LEVEL4":1, "CUST_LEVEL5":1})
  

• 用batch_size分批读取：别一次性把500万条数据全拉到内存里，给游标加batch_size参数，每次从Mongo拉取固定数量的文档，既减少内存压力，也避免单次请求超时：

  cursor = coll.find(...).batch_size(10000)  # 每次拉1万条，可根据内存情况调整
  

二、优化游标转DataFrame的过程 #

这一步的核心是避免不必要的内存开销，用Pandas原生的优化方法：

• 直接用from_records处理游标：别自己手动循环游标拼列表再转DataFrame，pd.DataFrame.from_records()可以直接接收PyMongo的游标，内部做了优化，速度比手动拼接快很多：

  import pandas as pd
  
  # 前面的查询游标
  cursor = coll.find({}, {"CUST_LEVEL1":1, "CUST_LEVEL2":1, "CUST_LEVEL3":1, "CUST_LEVEL4":1, "CUST_LEVEL5":1, "_id":0}).batch_size(10000)
  df = pd.DataFrame.from_records(cursor)
  

• 超大内存压力下的分批合并：如果你的机器内存不够（比如8GB以下），可以把数据分成多个小批次转成DataFrame，再合并成最终的大表：

  df_list = []
  batch_size = 50000
  cursor = coll.find(...).batch_size(batch_size)
  
  while True:
      # 每次拿一个批次的文档
      batch = list(cursor.next_batch())
      if not batch:
          break
      # 转成小DataFrame加入列表
      df_list.append(pd.DataFrame(batch))
      print(f"已处理 {len(df_list)*batch_size} 条数据")
  
  # 合并所有小DataFrame
  df = pd.concat(df_list, ignore_index=True)
  

三、进阶优化：让MongoDB多干活，Python少受累 #

如果需要对数据做预处理（比如分组、统计），尽量把计算逻辑放到Mongo端的聚合管道里，减少传到Python的数据量：
比如你想统计每个CUST_LEVEL1下的CUST_LEVEL2数量，直接用聚合管道完成：

pipeline = [
    {"$group": {"_id": "$CUST_LEVEL1", "level2_list": {"$addToSet": "$CUST_LEVEL2"}}},
    {"$project": {"_id": 0, "level1": "$_id", "level2_count": {"$size": "$level2_list"}}}
]
cursor = coll.aggregate(pipeline)
df = pd.DataFrame.from_records(cursor)

这样传到Python的只有聚合后的结果，数据量可能从几百万条降到几条，速度直接起飞。

另外，转成DataFrame后，可以把重复率高的字符串字段转成category类型，大幅减少内存占用：

for col in df.columns:
    df[col] = df[col].astype("category")

四、最后提几个环境配置建议 #

• 如果MongoDB服务器和Python程序不在同一台机器，尽量放到同一局域网或者本地，减少网络延迟；
• 给MongoDB调整wiredTiger缓存大小（比如设为机器内存的50%），提升查询性能；
• 给Python进程分配足够的内存，避免频繁的垃圾回收拖慢速度。

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