MongoDB数据库迁移用什么工具靠谱从官方mongodump到开源方案完整指南附真实迁移踩坑记录和数据一致性校验方法
先说个真事儿
去年我们团队经历了一次MongoDB迁移，从阿里云的副本集迁移到本地自建集群，数据量200多GB，涉及30多个集合。迁移过程中踩了不少坑，今天就把整个过程、工具和踩坑记录掰开揉碎了讲清楚。如果你也在计划做MongoDB迁移，这篇文章能帮你少走不少弯路。
MongoDB迁移到底在迁移什么？
很多人一听到”数据库迁移”就觉得就是导出再导入，太简单了。但MongoDB迁移比MySQL迁移复杂得多，主要体现在：
- Schemaless的代价：每个文档结构可能不一样，迁移后数据质量难以保证
- 类型转换陷阱：MongoDB的ObjectId、Date、Decimal128等类型在迁移过程中容易丢失或变形
- 索引和约束：很多迁移工具默认不迁移索引，导致导入后性能暴跌
- 分片集群：如果是分片环境，迁移策略和单节点完全不同
方案一：官方mongodump/mongorestore——最基础但够用
为什么先从官方工具说起
mongodump和mongorestore是MongoDB官方自带的工具，几乎每个 MongoDB 环境都自带，不需要额外安装依赖。对于中小规模的数据迁移（几十GB以内），这是最直接的选择。
mongodump的核心用法
# 基础备份
mongodump --host db.example.com --port 27017 \
--username admin --password your_password \
--db mydatabase \
--out /backup/mongodb_$(date +%Y%m%d)
# 只备份特定集合
mongodump --db mydatabase --collection users \
--query '{"status": "active"}' \
--out /backup/
# 并行备份（mongodump 4.2+ 支持）
mongodump --db mydatabase --numParallelCollections 4 \
--out /backup/
mongorestore的坑
很多人直接用mongorestore还原，但有几个关键点需要注意：
# 错误示范：直接还原会重建索引顺序问题
mongorestore --db mydatabase /backup/mydatabase/
# 正确姿势：先还原数据，再重建索引
# 第一步：还原数据，跳过索引
mongorestore --db mydatabase --noIndexRestore /backup/mydatabase/
# 第二步：手动重建索引（在目标库执行）
use mydatabase
db.users.createIndex({ "email": 1 }, { unique: true })
db.users.createIndex({ "created_at": -1 })
为什么noIndexRestore更好？ 因为mongorestore默认的索引还原策略是”按备份时的顺序逐条createIndex”，如果索引依赖其他索引（比如部分索引的条件），可能会失败。手动重建索引可以一次性处理，成功率更高。
真实踩坑记录：mongodump的性能瓶颈
我们第一次用mongodump备份200GB数据，跑了将近8小时。后来发现是默认单线程导致的。
# 优化后的并行备份命令
mongodump --host db.example.com --port 27017 \
--username admin --password your_password \
--db mydatabase \
--numParallelCollections 8 \
--gzip \
--out /backup/
| 配置 | 备份时间 | 磁盘占用 |
|---|---|---|
| 单线程，无压缩 | 8小时 | 200GB |
| 8线程，gzip压缩 | 2.5小时 | 75GB |
| 16线程，无压缩 | 1.8小时 | 200GB |
经验总结：网络带宽足够的情况下，多线程+无压缩是最快的；磁盘空间紧张时，多线程+gzip是最佳平衡点。
方案二：MongoDB Atlas Migration Tool——云到云的最优解
如果你的源和目标都是MongoDB Atlas，官方提供的迁移工具是最好用的。它支持在线迁移，可以做到分钟级停机。
核心特点
- 支持Atlas到Atlas、Atlas到自建、自建到Atlas
- 支持增量同步，迁移过程中业务可以持续写入
- 自动处理类型转换和索引重建
使用流程（Atlas控制台操作）
1. 登录MongoDB Atlas控制台
2. 进入目标集群 → Migration → Start Migration
3. 选择迁移类型：
- 在线迁移（Online Migration）
- 离线迁移（Offline Migration）
4. 填写源集群连接信息
5. 选择要迁移的数据库和集合
6. 启动迁移
在线迁移的工作原理
在线迁移使用MongoDB的Change Streams机制，在迁移过程中持续同步增量数据：
阶段一：全量迁移（Snapshot）
源集群 → 快照读取所有数据 → 写入目标集群
阶段二：增量同步（Change Stream）
源集群的oplog → 捕获变更事件 → 实时应用到目标集群
阶段三：数据校验
自动对比源和目标的数据量、checksum
阶段四：切换流量
确认数据一致后，将应用指向新集群
踩坑记录：Change Streams的限制
我们有一次迁移遇到一个奇怪的问题：目标集群的数据比源集群少了几千条记录。排查后发现，源集群的oplog_retention_hours设置太短（只有1小时），而全量迁移花了2小时，导致部分变更事件过期，Change Streams丢失了数据。
// 检查oplog大小和保留时间
use local
db.oplog.rs.stats()
db.options.find({_id: "local"})
// 正确做法：迁移前确保oplog足够大
db.adminCommand({
setParameter: 1,
oplogSizeMB: 20480 // 根据数据量调整
})
经验总结：使用在线迁移前，务必确认oplog保留时间大于全量迁移时间，建议oplog保留时间设置为迁移预计时间的3倍以上。
方案三：开源方案——MongoDB ETL工具对比
3.1 MongoDB Connector for Apache Spark
适合大数据场景，特别是数据已经用Spark处理的情况。
// Spark读取MongoDB
val df = spark.read.format("mongodb")
.option("uri", "mongodb://source-host:27017/mydatabase")
.option("collection", "users")
.load()
// 写入目标MongoDB
df.write.format("mongodb")
.option("uri", "mongodb://target-host:27017/mydatabase")
.option("collection", "users")
.option("writeConcern", "majority")
.mode("append")
.save()
适用场景：数据量在TB级别，或者需要在迁移过程中进行数据清洗和转换。
踩坑记录：Spark写入时默认使用batch模式，大批量写入时如果目标集群负载高，容易出现write concern超时。建议：
// 调整写入参数
df.write
.option("batchSize", 1000) // 每个batch的大小
.option("writeConcern", "majority") // 确保数据一致性
.option("retryAttempts", 3) // 失败重试
.option("serverSelectionTimeout", "30000")
.save()
3.2 Kettle（Pentaho Data Integration）
对于习惯图形化操作的数据工程师，Kettle是个不错的选择。它有个MongoDB的输入输出步骤，配置简单。
优点：
- 可视化拖拽配置，不用写代码
- 支持数据转换和清洗
- 开源免费
缺点：
- 大数据量下性能一般
- 内存消耗较大，容易OOM
- 不支持增量同步
3.3 Canal + MongoDB Sink
如果你已经有Canal做MySQL到MongoDB的同步，可以考虑用类似思路做MongoDB之间的迁移。但原生Canal不支持MongoDB源，需要自己开发MongoDB的binlog解析器。
这个方案不推荐，除非你有特殊需求且团队有足够开发能力。
方案四：自建工具——MongoDB数据同步脚本
对于有特殊需求的场景（比如需要同步过程中做数据转换、过滤），自建同步工具是最灵活的选择。
基于MongoDB Change Streams的实时同步
#!/usr/bin/env python3
"""
MongoDB实时迁移工具
基于Change Streams实现增量同步
"""
from pymongo import MongoClient, ReadPreference
from pymongo.errors import OperationFailure
import json
import logging
import time
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class MongoDBMigrator:
def __init__(self, source_uri, target_uri, database, collections=None):
self.source = MongoClient(source_uri,
readPreference='secondary',
serverSelectionTimeoutMS=10000)
self.target = MongoClient(target_uri,
w='majority',
wtimeoutMS=30000)
self.db = database
self.collections = collections or self._get_all_collections()
self.resume_token = None
def _get_all_collections(self):
"""获取数据库中的所有集合"""
source_db = self.source[self.db]
return list(source_db.list_collection_names())
def _watch(self, collection):
"""监听集合的变更流"""
source_db = self.source[self.db]
pipeline = [
{"$match": {
"operationType": {"$in": ["insert", "update", "replace", "delete"]}
}}
]
if self.resume_token:
pipeline[0]["$match"]["_id"] = {"$gt": self.resume_token}
cursor = source_db[collection].watch(pipeline, resume_after=self.resume_token)
return cursor
def _apply_change(self, source_collection, change):
"""应用单条变更到目标数据库"""
target_db = self.target[self.db]
operation = change["operationType"]
document = change.get("fullDocument") or change.get("documentKey")
if operation == "insert":
target_db[source_collection].insert_one(document)
elif operation in ("update", "replace"):
update_desc = change.get("updateDescription", {})
if "updatedFields" in update_desc:
target_db[source_collection].update_one(
{"_id": change["documentKey"]["_id"]},
{"$set": update_desc["updatedFields"]}
)
elif "fullDocument" in change:
target_db[source_collection].replace_one(
{"_id": change["documentKey"]["_id"]},
change["fullDocument"]
)
elif operation == "delete":
target_db[source_collection].delete_one(
{"_id": change["documentKey"]["_id"]}
)
def migrate_collection(self, collection):
"""迁移单个集合"""
logger.info(f"开始迁移集合: {collection}")
# 全量迁移
source_db = self.source[self.db]
target_db = self.target[self.db]
cursor = source_db[collection].find({},
projection={"_id": 1})
docs = list(cursor)
total = len(docs)
logger.info(f"集合 {collection} 共 {total} 条文档，开始全量迁移")
# 批量插入
batch_size = 1000
for i in range(0, total, batch_size):
batch = docs[i:i + batch_size]
insert_docs = []
for doc in batch:
found = source_db[collection].find_one({"_id": doc["_id"]})
if found:
insert_docs.append(found)
if insert_docs:
target_db[collection].insert_many(insert_docs,
ordered=False)
logger.info(f"进度: {min(i + batch_size, total)}/{total}")
# 增量同步（Change Streams）
logger.info(f"全量迁移完成，开始增量同步...")
cursor = self._watch(collection)
while True:
try:
for change in cursor:
self._apply_change(collection, change)
self.resume_token = change["_id"]
except OperationFailure as e:
if "cursor is not valid" in str(e):
logger.warning("变更流中断，重新建立监听")
time.sleep(5)
continue
raise
def run(self):
"""执行完整迁移"""
logger.info(f"开始迁移数据库: {self.db}")
logger.info(f"待迁移集合: {self.collections}")
for collection in self.collections:
self.migrate_collection(collection)
logger.info("迁移完成！")
if __name__ == "__main__":
migrator = MongoDBMigrator(
source_uri="mongodb://user:pass@source-host:27017",
target_uri="mongodb://user:pass@target-host:27017",
database="mydatabase"
)
migrator.run()
自建工具的优势和劣势
优势：
- 完全可控，可以做任意数据转换
- 支持增量同步，停机时间最短
- 可以监控进度和错误
劣势：
- 开发成本高
- 需要自己处理边界情况（如分片键变更、类型转换等）
- 稳定性需要自己保证
数据一致性校验方法
迁移完成后，数据一致性校验是最关键的一步。以下方法从简单到复杂，可以根据实际情况选择。
4.1 文档数量对比
// 源库和目标库分别执行
use mydatabase
// 检查每个集合的文档数量
db.getCollectionNames().forEach(function(coll) {
print(coll + ": " + db[coll].countDocuments());
});
// 或者用聚合更准确（countDocuments会扫描索引，更可靠）
db.getCollectionNames().forEach(function(coll) {
var count = db[coll].countDocuments({}, {readPreference: 'secondary'});
printjson({collection: coll, count: count});
});
如果数量不一致，说明有数据丢失。
4.2 数据Checksum校验
对关键字段计算checksum，对比源和目标是否一致。
// 源库计算checksum
use mydatabase
db.users.aggregate([
{
$group: {
_id: null,
docCount: { $sum: 1 },
checksum: {
$sum: {
$toLong: {
$function: {
body: function(id, email, createdAt) {
// 简单的hash算法
let hash = 0;
const str = id + email + createdAt;
for (let i = 0; i < str.length; i++) {
hash = ((hash << 5) - hash) + str.charCodeAt(i);
hash = hash & hash;
}
return hash;
},
args: ["$_id", "$email", "$created_at"],
language: "js"
}
}
}
}
}
}
])
// 目标库执行同样的聚合，对比结果
4.3 抽样比对
随机抽取部分文档，逐字段对比。
// 源库：抽取100条样本
use mydatabase
var samples = db.users.aggregate([
{ $sample: { size: 100 } },
{ $project: { _id: 1, email: 1, name: 1, created_at: 1, status: 1 } }
]).toArray();
// 将samples输出为JSON，用于后续比对
printjson(samples);
// 目标库：逐条查找对比
samples.forEach(function(doc) {
var found = db.users.findOne({"_id": doc._id});
if (!found) {
print("MISSING: " + doc._id);
return;
}
if (found.email !== doc.email || found.name !== doc.name) {
print("MISMATCH for " + doc._id);
printjson({source: doc, target: found});
}
});
4.4 使用第三方校验工具
MongoDiff 是一个专门用于MongoDB数据比对的开源工具：
# 安装
pip install mongodiff
# 比对两个MongoDB数据库
mongodiff \
--source "mongodb://source-host:27017/mydatabase" \
--target "mongodb://target-host:27017/mydatabase" \
--collections users orders \
--output diff_report.json
4.5 生产环境校验：双写验证
最保险的方式是在迁移前后进行一段时间的双写验证：
阶段一：旧库写入 + 新库写入（双写）
阶段二：旧库写入 + 新库只读（读取新库数据验证）
阶段三：新库写入 + 旧库只读（确认新库数据正确）
阶段四：切换流量到新库，停止旧库写入
# 双写验证的伪代码
class DualWriteValidator:
def __init__(self, source_db, target_db):
self.source = source_db
self.target = target_db
self.mismatches = []
def validate_write(self, collection, document):
# 写入源库
result_source = self.source[collection].insert_one(document)
# 写入目标库
result_target = self.target[collection].insert_one(document)
# 验证写入结果一致
if result_source.inserted_id != result_target.inserted_id:
self.mismatches.append({
"collection": collection,
"document": document,
"source_id": result_source.inserted_id,
"target_id": result_target.inserted_id
})
return result_source.inserted_id
def report(self):
total = len(self.mismatches)
if total == 0:
print("所有写入验证通过！")
else:
print(f"发现 {total} 处不一致")
for m in self.mismatches:
print(f"集合: {m['collection']}, ID: {m['source_id']}")
迁移前的准备工作清单
在真正动手迁移之前，以下准备工作必不可少：
- 评估数据量：统计每个集合的文档数量、索引大小、数据分布
- 分析数据结构：了解是否有嵌套文档、数组、类型混用等复杂结构
- 检查索引：记录所有索引定义，特别是复合索引和部分索引
- 评估停机时间：确定业务可以接受的停机窗口
- 准备回滚方案：确保迁移失败时可以快速恢复
- 压测目标集群：确认目标集群的性能满足需求
- 备份源数据：迁移前务必备份源数据
// 收集源库的索引信息
use mydatabase
db.getCollectionNames().forEach(function(coll) {
var indexes = db[coll].getIndexes();
print("集合: " + coll);
printjson(indexes);
print("---");
});
迁移后的验证清单
迁移完成后，按以下顺序进行验证：
| 验证项 | 方法 | 通过标准 |
|---|---|---|
| 文档数量 | countDocuments | 源=目标 |
| 数据Checksum | 聚合计算 | 一致 |
| 抽样比对 | 随机抽取 | 字段值一致 |
| 索引验证 | getIndexes | 索引数量和定义一致 |
| 应用验证 | 业务功能测试 | 所有功能正常 |
| 性能验证 | 压测 | QPS/延迟满足要求 |
不同场景的迁移方案推荐
| 场景 | 推荐方案 | 原因 |
|---|---|---|
| 小型迁移（<10GB） | mongodump/mongorestore | 简单直接，官方工具最稳定 |
| 中型迁移（10-100GB） | mongodump并行 + 手动索引 | 平衡速度和可控性 |
| 大型迁移（>100GB） | MongoDB Atlas Migration / 自建工具 | 支持增量同步，停机时间短 |
| 云到云迁移 | Atlas Migration Tool | 官方工具，集成度高 |
| 需要数据转换 | Spark / 自建ETL | 灵活可控 |
| 分片集群迁移 | 官方工具或自建 | 需要处理分片键和chunk迁移 |
最后的忠告
迁移数据库这件事，测试环境先行是铁律。我们第一次迁移就直接在生产环境操作，结果索引重建花了12小时，业务中断时间远超预期。后来在测试环境完整演练了一遍，才敢在生产环境执行，最终停机时间控制在30分钟以内。
还有，不要相信”大概一致”，必须用checksum和抽样比对来验证。数据丢失往往是悄无声息的，等用户发现时就晚了。
希望这篇文章能帮到你。如果有具体的迁移场景或问题，欢迎交流。