一、為什么需要文檔型數據庫？

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數據庫操作詳解