Python爬虫（10）Python数据存储实战：基于pymongo的MongoDB开发深度指南

一、为什么需要文档型数据库？

1.1 数据存储的范式变革

在移动互联网与物联网时代，‌非结构化数据占比超过80%‌（IDC报告）。传统关系型数据库（如MySQL）的固定表结构难以应对以下场景：

• ‌动态字段需求‌：用户画像标签频繁增减
• ‌海量数据写入‌：物联网设备每秒万级数据写入
• ‌复杂嵌套结构‌：一篇电商商品信息包含多级评论、规格参数

‌MongoDB作为文档型数据库的代表‌，采用BSON（Binary JSON）格式存储数据，支持动态模式、水平扩展和地理空间查询，成为大数据场景的核心基础设施。

1.2 pymongo的核心优势

作为MongoDB官方Python驱动，pymongo提供：

• ‌原生BSON支持‌：无缝处理Python字典与BSON的转换
• ‌连接池管理‌：自动管理TCP连接复用
• ‌聚合管道封装‌：支持复杂数据分析操作
• ‌完善的API‌：覆盖索引管理、副本集操作等高级功能

二、pymongo核心操作全解析

2.1 环境准备

# 安装MongoDB社区版（以Ubuntu为例）
wget -qO - https://www.mongodb.org/static/pgp/server-6.0.asc | sudo apt-key add -
echo "deb [ arch=amd64,arm64 ] https://repo.mongodb.org/apt/ubuntu focal/mongodb-org/6.0 multiverse" | sudo tee /etc/apt/sources.list.d/mongodb-org-6.0.list
sudo apt-get update
sudo apt-get install -y mongodb-org# 安装pymongo
pip install pymongo

2.2 数据库连接与CRUD操作

from pymongo import MongoClient
from pymongo.errors import ConnectionFailure# 建立连接（默认连接池大小100）
client = MongoClient(host="localhost",port=27017,username="admin",  # 启用身份验证时必填password="securepassword",authSource="admin"
)try:# 心跳检测client.admin.command('ping')print("Successfully connected to MongoDB!")
except ConnectionFailure:print("Server not available")# 选择数据库与集合（自动懒创建）
db = client["ecommerce"]
products_col = db["products"]# 插入文档（自动生成_id）
product_data = {"name": "Wireless Mouse","price": 49.99,"tags": ["electronics", "computer"],"stock": {"warehouse_A": 100, "warehouse_B": 50},"last_modified": datetime.now()
}
insert_result = products_col.insert_one(product_data)
print(f"Inserted ID: {insert_result.inserted_id}")# 查询文档（支持嵌套查询）
query = {"price": {"$lt": 60}, "tags": "electronics"}
projection = {"name": 1, "price": 1}  # 类似SQL SELECT
cursor = products_col.find(query, projection).limit(5)
for doc in cursor:print(doc)# 更新文档（原子操作）
update_filter = {"name": "Wireless Mouse"}
update_data = {"$inc": {"stock.warehouse_A": -10}, "$set": {"last_modified": datetime.now()}}
update_result = products_col.update_one(update_filter, update_data)
print(f"Modified count: {update_result.modified_count}")# 删除文档
delete_result = products_col.delete_many({"price": {"$gt": 200}})
print(f"Deleted count: {delete_result.deleted_count}")

2.3 聚合管道实战

# 统计各仓库库存总量
pipeline = [{"$unwind": "$stock"},  # 展开嵌套文档{"$group": {"_id": "$stock.warehouse","total_stock": {"$sum": "$stock.quantity"}}},{"$sort": {"total_stock": -1}}
]
results = products_col.aggregate(pipeline)
for res in results:print(f"Warehouse {res['_id']}: {res['total_stock']} units")

2.4 分批次插入百万级数据（进阶）

from pymongo import MongoClient
from faker import Faker
import timeclient = MongoClient('mongodb://localhost:27017/')
db = client['bigdata']
collection = db['user_profiles']fake = Faker()
batch_size = 5000  # 分批次插入减少内存压力def generate_batch(batch_size):return [{"name": fake.name(),"email": fake.email(),"last_login": fake.date_time_this_year()} for _ in range(batch_size)]start_time = time.time()
for _ in range(200):  # 总数据量100万batch_data = generate_batch(batch_size)collection.insert_many(batch_data, ordered=False)  # 无序插入提升速度print(f"已插入 {(i+1)*batch_size} 条数据")print(f"总耗时: {time.time()-start_time:.2f}秒") 

2.5 分批次插入百万级数据（进阶）

# 分析电商订单数据（含嵌套结构）
pipeline = [{"$unwind": "$items"},  # 展开订单中的商品数组{"$match": {"status": "completed"}},  # 筛选已完成订单{"$group": {"_id": "$items.category","total_sales": {"$sum": "$items.price"},"avg_quantity": {"$avg": "$items.quantity"},"top_product": {"$max": "$items.name"}}},{"$sort": {"total_sales": -1}},{"$limit": 10}
]orders_col = db["orders"]
results = orders_col.aggregate(pipeline)for res in results:print(f"品类 {res['_id']}: 销售额{res['total_sales']}元")

三、生产环境进阶配置

3.1 性能优化关键措施

# 创建索引（提升查询速度）
products_col.create_index([("name", pymongo.ASCENDING)], unique=True)
products_col.create_index([("price", pymongo.ASCENDING), ("tags", pymongo.ASCENDING)])# 批量写入提升吞吐量
bulk_ops = [pymongo.InsertOne({"name": "Keyboard", "price": 89.99}),pymongo.UpdateOne({"name": "Mouse"}, {"$set": {"price": 59.99}}),pymongo.DeleteOne({"name": "Earphones"})
]
results = products_col.bulk_write(bulk_ops)

3.2 高可用架构配置

# MongoDB副本集配置（3节点）
replication:replSetName: "rs0"members:- _id: 0, host: "mongo1:27017"- _id: 1, host: "mongo2:27017"- _id: 2, host: "mongo3:27017", arbiterOnly: true

3.3 安全加固方案

# 启用身份验证
use admin
db.createUser({user: "admin",pwd: "securepassword",roles: [ { role: "userAdminAnyDatabase", db: "admin" } ]
})# 配置网络加密
net:tls:mode: requireTLScertificateKeyFile: /etc/ssl/mongo.pem

四、总结与最佳实践

4.1 技术选型对比‌：

特性	MongoDB	MySQL
数据模型	动态文档	固定表结构
扩展方式	水平分片	垂直扩展
事务支持	4.0+版本支持	原生完善支持
适用场景	日志/用户行为数据	金融交易系统

4.2 性能优化原则‌：

1. ‌Working Set‌原则：确保常用数据能放入内存
2. ‌索引覆盖‌：通过组合索引避免回表查询
3. ‌分片键设计‌：选择高基数、易分散的字段

4‌.3 避坑指南‌：

• 避免文档无限制增长（推荐设置capped collection）
• 慎用$where操作符（导致全表扫描）
• 生产环境必须配置副本集与定期备份

Python爬虫相关文章（推荐）

Python爬虫介绍	Python爬虫（1）Python爬虫：从原理到实战，一文掌握数据采集核心技术
HTTP协议解析	Python爬虫（2）Python爬虫入门：从HTTP协议解析到豆瓣电影数据抓取实战
HTML核心技巧	Python爬虫（3）HTML核心技巧：从零掌握class与id选择器，精准定位网页元素
CSS核心机制	Python爬虫（4）CSS核心机制：全面解析选择器分类、用法与实战应用
静态页面抓取实战	Python爬虫（5）静态页面抓取实战：requests库请求头配置与反反爬策略详解
静态页面解析实战	Python爬虫（6）静态页面解析实战：BeautifulSoup与lxml（XPath）高效提取数据指南
Python数据存储实战 CSV文件	Python爬虫（7）Python数据存储实战：CSV文件读写与复杂数据处理指南
Python数据存储实战 JSON文件	Python爬虫（8）Python数据存储实战：JSON文件读写与复杂结构化数据处理指南
Python数据存储实战 MySQL数据库	Python爬虫（9）Python数据存储实战：基于pymysql的MySQL数据库操作详解