1. 当数据导入变成"慢动作"
去年我们团队接手了一个物联网设备日志分析项目，需要将每天500万条设备日志导入MongoDB。最初的方案是用Python脚本逐条插入，结果发现每小时只能处理3万条数据。运维同事盯着监控大屏上的龟速进度条，绝望得就像在等手机系统更新——明明进度条走完了，还要再等"最后三分钟"。
这里有个典型的反面教材：
# 错误示范：逐条插入（Python + PyMongo）
from pymongo import MongoClient
import time
client = MongoClient('mongodb://user:pass@1.1.1.1:27017')
db = client.iot_logs
collection = db.devices
def insert_logs(logs):
start = time.time()
for log in logs: # 逐个遍历
collection.insert_one(log) # 单文档插入
print(f"插入耗时：{time.time()-start:.2f}s")
# 测试10万条数据
logs = [{"device_id": i, "temp": 25.5} for i in range(100000)]
insert_logs(logs) # 实际测试耗时约85秒
这种写法就像用勺子运沙子，每次只能带一粒。当数据量达到百万级时，网络往返开销会吃掉大部分时间。更糟糕的是，每次插入都会产生写锁竞争，导致数据库像早高峰的地铁口一样拥堵。
2. 批量操作：从勺子到传送带
MongoDB的批量操作API就像是给数据装上了传送带。我们改造后的代码：
# 正确姿势：批量插入（Python + PyMongo）
from pymongo import MongoClient, InsertOne
import time
client = MongoClient('mongodb://user:pass@1.1.1.1:27017', maxPoolSize=50)
db = client.iot_logs
collection = db.devices
def bulk_insert(logs, batch_size=1000):
start = time.time()
operations = []
for i, log in enumerate(logs):
operations.append(InsertOne(log))
if (i+1) % batch_size == 0 or i == len(logs)-1:
collection.bulk_write(operations, ordered=False)
operations = []
print(f"批量插入耗时：{time.time()-start:.2f}s")
# 同样测试10万条数据
logs = [{"device_id": i, "temp": 25.5} for i in range(100000)]
bulk_insert(logs) # 耗时降至约2.7秒，提升31倍！
这里有几个关键点：
batch_size=1000：相当于传送带的装载量，太大可能导致内存溢出ordered=False：允许乱序执行，类似快递分拣流水线maxPoolSize=50：连接池扩容，避免请求在客户端排队
但批量操作不是银弹，我们在电商订单系统迁移时遇到过坑：当批量写入10万条用户评价时，由于未设置合适的写关注级别，导致部分写入失败后难以追溯。后来调整为：
collection.bulk_write(
operations,
ordered=False,
write_concern={"w": "majority", "wtimeout": 5000}
)
3. 网络调优：看不见的战场
有一次客户投诉数据导入速度波动大，我们排查三天后发现是办公网QOS策略在作祟。这提醒我们：网络环境就像高速公路，路况不好再好的车也跑不快。
3.1 连接池设置技巧
# 优化连接配置（Python + PyMongo）
client = MongoClient(
'mongodb://user:pass@1.1.1.1:27017',
maxPoolSize=100, # 连接池大小
minPoolSize=20, # 最小保活连接
socketTimeoutMS=30000, # 单次操作超时
connectTimeoutMS=5000, # 连接建立超时
waitQueueTimeoutMS=2000 # 获取连接等待时间
)
这组参数相当于：
- 保持20辆出租车随时待命（minPoolSize）
- 最多派出100辆同时运营（maxPoolSize）
- 司机接单后30秒没接到乘客就取消（socketTimeoutMS）
- 叫车等待超过2秒就换平台（waitQueueTimeoutMS）
3.2 压缩协议：看不见的加速器
当我们在跨国数据中心同步数据时，启用压缩协议后传输效率提升40%：
client = MongoClient(
'mongodb://user:1.1.1.1:27017',
compressors="snappy,zlib", # 压缩算法优先级
zlibCompressionLevel=3 # 压缩级别
)
不同压缩算法的选择就像打包行李：
- Snappy：快速打包但箱子稍大（CPU占用低）
- Zlib：精心整理节省空间（压缩率高）
- 建议在带宽受限时使用zlib，CPU密集型场景用snappy
4. 关联技术：管道工的瑞士军刀
4.1 Change Stream监听优化
在处理实时数据流时，合理设置重试策略能避免网络抖动带来的中断：
with collection.watch(
[{"$match": {"operationType": "insert"}}],
maxAwaitTimeMS=30000,
batchSize=500,
full_document="updateLookup"
) as stream:
for change in stream:
process_change(change)
参数解读：
maxAwaitTimeMS：最长等待新数据的时间（类似外卖小哥等单时限）batchSize：每次拉取的最大事件数full_document：是否获取完整文档镜像
4.2 聚合管道预处理
对于需要即时统计的场景，在导入阶段进行预聚合：
# 设备状态统计预聚合（Python + PyMongo）
pipeline = [
{"$match": {"status": {"$exists": True}}},
{"$group": {
"_id": "$device_type",
"total": {"$sum": 1},
"online": {"$sum": {"$cond": [{"$eq": ["$status", "online"]}, 1, 0]}}
}},
{"$merge": {"into": "device_stats"}}
]
collection.aggregate(pipeline, allowDiskUse=True)
这个管道实现了：
- 筛选包含状态字段的文档
- 按设备类型分组统计总数和在线数
- 将结果合并到统计集合
allowDiskUse=True就像允许使用扩展内存，避免大数据集撑爆内存
5. 性能调优的"不可能三角"
经过多个项目的实战，我们总结出以下经验矩阵：
| 优化方向 | 适用场景 | 优点 | 注意事项 |
|---|---|---|---|
| 批量写入 | 初始数据迁移/定时ETL | 吞吐量高，网络开销小 | 需要合理设置批次大小 |
| 连接池优化 | 高并发持续写入 | 减少连接建立开销 | 需监控连接数防止服务端过载 |
| 压缩传输 | 跨数据中心同步 | 节省带宽成本 | 增加CPU消耗，需平衡压缩级别 |
| 预处理聚合 | 实时分析场景 | 降低查询延迟 | 增加写入复杂度，需维护一致性 |
记得某次在金融交易系统优化时，我们同时启用了批量写入和压缩协议，结果CPU使用率飙升触发告警。最后发现是zlib压缩级别设置过高，调整为3级后CPU占用下降60%而压缩率仅降低15%。
6. 写给技术选型者的备忘录
当你要进行大规模数据导入时，请记住这个检查清单：
【容量规划】
- 预估日均数据增长量 × 3（保留扩容空间）
- 索引内存占用不超过可用内存的60%
【写入策略】
- 批量大小从1000起步，逐步测试最优值
- 非事务场景优先使用无序写入
【网络配置】
- 跨机房部署必须开启压缩
- TCP KeepAlive保持时间建议120-300秒
【容错机制】
- 重试策略要包含指数退避
- 重要数据使用majority写关注
【监控指标】
- 关注oplog时间戳延迟
- 连接池等待时间超过1秒需要扩容
最后分享一个真实案例：某智能家居平台在采用优化方案后，将3000万设备数据的导入时间从14小时压缩到47分钟。他们的秘诀是：
- 使用5000条/批的批量写入
- 开启snappy压缩
- 调整writeConcern为w:1
- 在客户端实现断点续传机制
数据导入优化就像疏通城市交通，既需要拓宽道路（网络优化），也要提高车辆运力（批量操作），更得设计好交通规则（写入策略）。希望这些实战经验能让你下次面对数据洪流时，可以淡定地打开正确的水闸。