如何使用Python并行从MongoDB导入数据？

想要并行从MongoDB导入数据？这里有个实用的解决方案：先扫描目标集合（假设里面总共有1000条数据），接着把数据拆分成每批100条的小批次，最后将所有批次的数据合并成完整的1000条数据集。

先给你一个从MongoDB向Python导入数据的基础连接工具代码，我们可以基于这个扩展并行逻辑：

import pandas as pd
from pymongo import MongoClient

def _connect_mongo(host, port, username, password, db):
    """ 用于建立MongoDB连接的工具函数 """
    if username and password:
        mongo_uri = 'mongodb://%s:%s@%s:%s/%s' % (username, password, host, port, db)
        client = MongoClient(mongo_uri)
    else:
        client = MongoClient(host, port)
    return client[db]

接下来是具体的并行实现方案，我用Python的concurrent.futures库来实现多线程并行导入，这样既能提升效率，又不会太复杂：

from concurrent.futures import ThreadPoolExecutor

def fetch_batch(collection, skip, limit):
    """ 获取指定批次的数据 """
    return list(collection.find().skip(skip).limit(limit))

def parallel_import_mongo(db_name, collection_name, batch_size=100, host='localhost', port=27017, username=None, password=None):
    # 先建立MongoDB连接，拿到目标集合
    db = _connect_mongo(host, port, username, password, db_name)
    collection = db[collection_name]
    
    # 计算总数据量和需要的批次数量
    total_count = collection.count_documents({})
    total_batches = (total_count + batch_size - 1) // batch_size  # 向上取整计算批次
    
    # 用线程池并行获取每个批次的数据
    with ThreadPoolExecutor() as executor:
        futures = []
        for batch_idx in range(total_batches):
            skip_num = batch_idx * batch_size
            futures.append(executor.submit(fetch_batch, collection, skip_num, batch_size))
        
        # 收集所有批次的结果并合并
        all_data = []
        for future in futures:
            all_data.extend(future.result())
    
    # 转换成DataFrame（如果需要用pandas处理的话）
    return pd.DataFrame(all_data)

# 调用示例：替换成你的数据库和集合名称
final_df = parallel_import_mongo('your_database', 'your_collection', batch_size=100)

这里还有几个需要注意的点：

• 线程池的大小可以根据你的服务器配置调整，默认会根据CPU核心数设置，不用盲目调大，避免给MongoDB造成过大压力
• 如果你的数据量特别大，也可以考虑用多进程，但要注意每个进程需要单独建立MongoDB连接，不要共享连接对象
• batch_size的大小可以根据内存情况调整，内存充足的话可以适当调大，减少批次数量

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