如何使用 mongoexport mongoimport 完成 MongoDB 数据库平滑迁移与数据完整性校验指南

想象一下，你手里拿着一个装满珍贵照片的旧相册，现在要搬到新房子。你不能直接把相册扔进卡车，那样照片会散、会皱、甚至会丢。你得一页一页小心地整理、核对，再装进一个更结实的相册里。数据库迁移也是这个道理，尤其是用 mongoexport 和 mongoimport 这两把“瑞士军刀”——它们简单、直接，但要是用错了姿势，数据就可能“皱巴巴”甚至“丢页”。

今天，我们不讲枯燥的理论，就用一个真实的场景，带你走完从导出到导入，再到校验的完整流程。你会发现，其实这事儿没那么可怕。

第一步：理解这两把工具的本质

mongoexport 把 MongoDB 里的文档变成 JSON 或 CSV 文件。mongoimport 则反过来，把文件里的数据塞回 MongoDB。

关键点在于：JSON 是最推荐、最安全的格式。CSV 看似方便，但它会丢失数据类型——比如 ObjectId 会变成字符串，日期会变成字符串，NumberInt 和 NumberLong 可能会混为一谈。这些“隐形”的类型变化，在查询时不会报错，但在业务逻辑里可能引发致命 bug。

所以，除非你有特殊理由（比如数据量极大且结构极其简单），否则一律用 JSON。

第二步：导出前的准备——别急着敲命令

在源服务器上，先确认几件事：

1. 检查空间 导出文件会占用磁盘空间。假设你的数据库是 50GB，压缩后的 JSON 可能接近 30-40GB。确保目标位置有足够的空间。可以用 df -h 看看。

2. 确定导出范围 你是要导整个库？还是某个集合？还是某个查询条件下的子集？ mongoexport 支持 --query 参数，这意味着你可以只导出“近一年”的数据，而不是全部。这对大数据量迁移特别有用。

3. 停止写入（可选但推荐） 如果业务允许，在导出前让源库停止写入，或者至少停止对目标集合的写入。否则，你可能会遇到“导出一半，有新数据写入，导致数据不一致”的情况。对于 7x24 小时运行的服务，这很难做到，我们后面会讲如何处理这种情况。

4. 创建备份 虽然这只是导出，不是备份，但养成习惯：在源库上先做一次全量备份（比如用 mongodump）。万一导出过程出问题，你还有退路。

第三步：执行导出——命令详解

下面这个命令是我在实际项目中常用的版本，每个参数都有它的意义：

mongoexport \
  --uri="mongodb://username:password@source-host:27017" \
  --db my_database \
  --collection my_collection \
  --out /backup/mongo_exports/my_collection.json \
  --jsonFormat compact \
  --type json \
  --query '{"createdAt": {"$gte": "2023-01-01T00:00:00Z"}}' \
  --batchSize 10000 \
  --noIndexRestore

--uri：新版 mongotool 推荐用 URI 格式，代替旧的 --host、--port、--username、--password。更简洁，也支持认证数据库。
--out：指定输出文件路径。注意，mongoexport 会自动在文件名前加上 .json 后缀（如果你没在命令里写的话）。所以 --out my_collection 和 --out my_collection.json 效果一样，都会生成 my_collection.json。
--jsonFormat compact：紧凑格式，每行一个 JSON 文档，便于 mongoimport 直接读取，也节省空间。如果不加，默认是pretty格式，可读性好但文件巨大，导入速度慢。
--query：可选。如果你只想导出部分数据，用这个。支持完整的 MongoDB 查询语法。
--batchSize 10000：每次从数据库读取的文档数。调大可以加速导出，但会占用更多内存。如果内存紧张，调小。
--noIndexRestore：这个参数只在导入时有意义，导出时忽略。

多集合导出技巧 如果你有几十个集合要导，一个一个敲命令太慢了。写个脚本：

#!/bin/bash
DB="my_database"
URI="mongodb://user:pass@source-host:27017"
OUT_DIR="/backup/mongo_exports"

# 获取所有集合名称
COLLECTIONS=$(mongo "$URI/$DB" --quiet --eval "db.getCollectionNames().join('\n')" | tr -d '[]"')

for COL in $COLLECTIONS; do
  echo "Exporting $COL ..."
  mongoexport \
    --uri="$URI" \
    --db="$DB" \
    --collection="$COL" \
    --out="$OUT_DIR/$COL.json" \
    --jsonFormat compact \
    --type json
  echo "Finished $COL"
done

这个脚本会遍历所有集合，逐一导出。日志清晰，出错也好排查。

性能优化 如果数据量很大（超过 100GB），导出可能很慢。几个建议：

并行导出：用 GNU parallel 或 xargs 并行执行多个 mongoexport 进程。但要注意不要压垮源库。
使用 --projection：如果某些字段不需要，只导出需要的字段，可以显著减小文件体积。
SSD 磁盘：导出到 SSD 比机械盘快得多。

第四步：传输文件——别用 scp 传一个 50GB 的文件

导出完成后，你需要把 JSON 文件从源服务器传到目标服务器。scp 可以，但如果网络不稳定，传一半断了，你得重头再来。

推荐方案：rsync

rsync -avz --progress /backup/mongo_exports/ user@target-host:/backup/mongo_exports/

-a：归档模式，保留权限、时间等。
-v：显示进度。
-z：压缩传输，节省带宽。
--progress：显示每个文件的传输进度。

rsync 的好处是断点续传。如果传了一半断了，重新运行同一命令，它只会传输未完成的部分。

如果文件太大（比如几百 GB） 考虑分卷压缩：

# 导出时直接分卷
tar czf - /backup/mongo_exports/ | split -b 10G - /backup/mongo_exports_part_

# 传输后合并
cat /backup/mongo_exports_part_* | tar xzf - -C /target/backup/mongo_exports/

这样每个分卷 10GB，即使某个分卷传输失败，只需重传那个分卷。

第五步：导入前准备——目标库同样重要

在目标服务器上，先做这些：

1. 确认 MongoDB 版本兼容性 mongoexport 导出的数据，理论上可以导入到相同或更高版本的 MongoDB。不能从高版本导出的 JSON 导入到低版本，因为低版本可能不认识新的数据类型或操作符。

检查版本：

mongod --version
mongoimport --version

2. 创建数据库和集合（可选） mongoimport 会自动创建数据库和集合，所以这一步不是必须的。但提前创建可以让你在导入前就发现权限问题。

3. 停止写入（如果可能） 和导出一样，如果业务允许，导入期间停止对目标集合的写入。否则，你导入的数据可能会被新写入的数据覆盖，或者新写入的数据在导入期间被跳过。

4. 检查磁盘空间 确保目标服务器有足够的空间存放导入后的数据。JSON 文件解压后通常会比原始 BSON 文件大一些。

第六步：执行导入——命令详解

和导出类似，导入也有讲究：

mongoimport \
  --uri="mongodb://username:password@target-host:27017" \
  --db my_database \
  --collection my_collection \
  --file /backup/mongo_exports/my_collection.json \
  --jsonFormat compact \
  --type json \
  --batchSize 10000 \
  --drop

--file：指定要导入的文件。
--drop：关键参数！ 如果目标集合已存在，--drop 会先删除现有数据，再导入新数据。这能避免数据重复。但请注意：这会永久删除目标集合的所有数据。如果目标集合还有其他业务在用，慎用。
--mode insert（默认）：直接插入。如果文档 _id 已存在，会报错并跳过。
--mode upsert：如果文档 _id 存在则更新，不存在则插入。适合增量同步场景。

多集合导入脚本

#!/bin/bash
DB="my_database"
URI="mongodb://user:pass@target-host:27017"
IMPORT_DIR="/backup/mongo_exports"

for FILE in "$IMPORT_DIR"/*.json; do
  COL=$(basename "$FILE" .json)
  echo "Importing $COL ..."
  mongoimport \
    --uri="$URI" \
    --db="$DB" \
    --collection="$COL" \
    --file="$FILE" \
    --jsonFormat compact \
    --type json \
    --batchSize 10000 \
    --drop
  echo "Finished $COL"
done

性能优化

并行导入：和导出一样，可以用 parallel 并行导入多个集合。但要注意目标库的负载。
禁用索引：导入完成后，再创建索引。导入过程中创建索引会拖慢速度。可以在导入时用 --noIndexRestore（默认行为），然后在导入后用 db.collection.createIndex() 手动创建索引。
调整 writeConcern：默认是 w: 1（主库写入即成功）。如果追求速度，可以临时设为 w: 0，但会失去数据一致性保证。不推荐生产环境用。

第七步：数据完整性校验——最关键的一步

导入完成后，千万别急着切换业务！你需要确认数据真的“完美迁移”了。

7.1 文档数量比对

最简单的方法：比对源库和目标库的文档数量。

# 源库
SOURCE_COUNT=$(mongo "mongodb://user:pass@source-host:27017/my_database" --quiet --eval "db.my_collection.countDocuments()")

# 目标库
TARGET_COUNT=$(mongo "mongodb://user:pass@target-host:27017/my_database" --quiet --eval "db.my_collection.countDocuments()")

echo "Source: $SOURCE_COUNT, Target: $TARGET_COUNT"

如果数量一致，说明没有明显的丢失或重复。但注意：数量一致不代表数据完全一致。可能有些文档被修改了，或者有些字段的值变了。

7.2 抽样比对

随机抽取几个文档，在源库和目标库之间比对。可以用 _id 来定位。

# 获取一个随机 _id
RANDOM_ID=$(mongo "mongodb://user:pass@source-host:27017/my_database" --quiet --eval "db.my_collection.aggregate([{$sample: {size: 1}}]).toArray()[0]._id.toString()")

# 从源库查询
SOURCE_DOC=$(mongo "mongodb://user:pass@source-host:27017/my_database" --quiet --eval "printjson(db.my_collection.findOne({_id: ObjectId('$RANDOM_ID')}))")

# 从目标库查询
TARGET_DOC=$(mongo "mongodb://user:pass@target-host:27017/my_database" --quiet --eval "printjson(db.my_collection.findOne({_id: ObjectId('$RANDOM_ID')}))")

# 简单比对（忽略 _id 本身，因为导入后可能略有不同）
if [ "$SOURCE_DOC" == "$TARGET_DOC" ]; then
  echo "Match for $RANDOM_ID"
else
  echo "Mismatch for $RANDOM_ID"
  echo "Source:"
  echo "$SOURCE_DOC"
  echo "Target:"
  echo "$TARGET_DOC"
fi

这个脚本只比对一个文档。你可以写个循环，比对多个文档。

7.3 哈希比对——更彻底的校验

如果你想要更彻底的校验，可以计算每个文档的哈希值，然后比对哈希值的集合。

# 源库：计算每个文档的哈希，输出哈希列表
mongo "mongodb://user:pass@source-host:27017/my_database" --quiet --eval "
  db.my_collection.find({}, {_id: 1, data: 1}).forEach(function(doc) {
    var hash = CryptoJS.SHA256(JSON.stringify(doc)).toString();
    print(hash);
  });
" > source_hashes.txt

# 目标库：同样计算哈希
mongo "mongodb://user:pass@target-host:27017/my_database" --quiet --eval "
  db.my_collection.find({}, {_id: 1, data: 1}).forEach(function(doc) {
    var hash = CryptoJS.SHA256(JSON.stringify(doc)).toString();
    print(hash);
  });
" > target_hashes.txt

# 比对哈希列表
sort source_hashes.txt > source_hashes_sorted.txt
sort target_hashes.txt > target_hashes_sorted.txt
diff source_hashes_sorted.txt target_hashes_sorted.txt

如果 diff 没有输出，说明哈希列表完全一致，数据完全相同。

注意：这个方法假设你只关心文档内容，不关心字段顺序。因为 JSON.stringify 对字段顺序敏感。如果源库和目标库的字段顺序不同，哈希会不同，但数据其实是相同的。要避免这个问题，可以在哈希前对文档进行排序。

// 排序后的哈希
function sortedHash(doc) {
  var sortedKeys = Object.keys(doc).sort();
  var sortedDoc = {};
  sortedKeys.forEach(function(key) {
    sortedDoc[key] = doc[key];
  });
  return CryptoJS.SHA256(JSON.stringify(sortedDoc)).toString();
}

7.4 业务逻辑校验

最后，也是最容易被忽视的一步：业务逻辑校验。

举个例子：如果你的系统里有“订单总金额”字段，迁移后，你可以跑一个 SQL 风格的聚合查询，比对总金额是否一致。

# 源库
SOURCE_TOTAL=$(mongo "mongodb://user:pass@source-host:27017/my_database" --quiet --eval "
  db.orders.aggregate([
    { \$match: { createdAt: { \$gte: ISODate('2023-01-01') } } },
    { \$group: { _id: null, total: { \$sum: '\$amount' } } }
  ]).toArray()[0].total
")

# 目标库
TARGET_TOTAL=$(mongo "mongodb://user:pass@target-host:27017/my_database" --quiet --eval "
  db.orders.aggregate([
    { \$match: { createdAt: { \$gte: ISODate('2023-01-01') } } },
    { \$group: { _id: null, total: { \$sum: '\$amount' } } }
  ]).toArray()[0].total
")

echo "Source total: $SOURCE_TOTAL, Target total: $TARGET_TOTAL"

如果总金额一致，说明数据迁移在业务层面也是正确的。

第八步：处理持续写入——如何实现“平滑”迁移？

前面我们假设可以停止写入。但很多业务是 7x24 小时的，不能停机。怎么办？

方案一：双写 在迁移期间，业务同时写入源库和目标库。迁移完成后，只写目标库。这需要业务代码支持，改动较大。

方案二：增量同步

先全量导出导入（如前所述）。
然后，找出全量导入时间点之后的“增量数据”，再导出导入一次。
用查询条件 {"createdAt": {"$gte": "全量导入时间点"}} 来获取增量数据。

# 第一次全量导出
mongoexport --uri="mongodb://user:pass@source-host:27017" --db my_database --collection my_collection --out full.json --jsonFormat compact --type json

# 记录全量导入完成的时间点（假设是 2024-01-01T00:00:00Z）
INCREMENTAL_TIME="2024-01-01T00:00:00Z"

# 增量导出
mongoexport --uri="mongodb://user:pass@source-host:27017" --db my_database --collection my_collection --out incremental.json --jsonFormat compact --type json --query "{ \"createdAt\": { \"\$gte\": \"${INCREMENTAL_TIME}\" } }"

# 增量导入
mongoimport --uri="mongodb://user:pass@target-host:27017" --db my_database --collection my_collection --file incremental.json --jsonFormat compact --type json --mode upsert

这个方案有个前提：你的集合里必须有一个“更新时间”字段（比如 createdAt 或 updatedAt），并且这个字段在文档创建或更新时会被正确设置。

方案三：MongoDB Change Streams MongoDB 4.0+ 支持 Change Streams，可以实时捕获数据变更。你可以搭建一个同步管道，实时将源库的变更同步到目标库。但这比较复杂，需要额外的基础设施。

对于大多数场景，方案二（增量同步） 是最实用、最容易实现的。

第九步：常见问题与排查

问题1：导入时报“BSON size exceeds limit” 原因：某些文档太大，超过了 BSON 文档大小限制（默认 16MB）。解决：检查并拆分大文档，或者调大限制（不推荐）。

问题2：导入后数据类型变了 原因：用了 CSV 格式，或者 JSON 里没有明确类型。解决：确保用 JSON 格式，并且源库导出的 JSON 包含类型信息（mongoexport 默认会包含 $type

正文

如何使用 mongoexport mongoimport 完成 MongoDB 数据库平滑迁移与数据完整性校验指南

第一步：理解这两把工具的本质

第二步：导出前的准备——别急着敲命令

第三步：执行导出——命令详解

第四步：传输文件——别用 scp 传一个 50GB 的文件

第五步：导入前准备——目标库同样重要

第六步：执行导入——命令详解

第七步：数据完整性校验——最关键的一步

7.1 文档数量比对

7.2 抽样比对

7.3 哈希比对——更彻底的校验

7.4 业务逻辑校验

第八步：处理持续写入——如何实现“平滑”迁移？

第九步：常见问题与排查

相关阅读

轻松掌握MongoDB数据迁移技巧，告别手动操作烦恼，实现高效迁移解决方案

Fedora系统迁移：轻松步骤，安全过渡，小白也能轻松上手

轻松学会SQL数据迁移：一站式工具详解与实战案例分享

掌握MongoDB数据迁移，轻松跨平台操作全攻略

Fedora系统迁移：轻松转移，无忧升级，新手必看实用攻略

MongoDB数据库迁移：轻松实现跨平台迁移，高效迁移数据，确保业务无缝衔接

轻松迁移Fedora系统：一键操作，无忧升级，新手也轻松上手！

教你轻松上手：MongoDB数据迁移全攻略，从方案选择到实操步骤一网打尽

sqlite数据库迁移：轻松实现版本控制与数据同步，告别手动操作烦恼

轻松掌握SQL数据迁移：三款实用工具，轻松实现数据迁移无忧