Amazon DocumentDB 迁移指南 #

常见迁移场景：
├── 从自建MongoDB迁移
├── 从MongoDB Atlas迁移
├── 从其他云服务迁移
├── 从其他数据库迁移
└── 数据中心迁移

迁移方式：
├── 在线迁移
│   ├── 使用DMS
│   └── 最小停机时间
│
├── 离线迁移
│   ├── 导出导入
│   └── 需要停机
│
└── 混合迁移
    ├── 全量+增量
    └── 平滑迁移

迁移流程：
├── 1. 评估和规划
├── 2. 兼容性检查
├── 3. 环境准备
├── 4. 数据迁移
├── 5. 应用适配
├── 6. 测试验证
├── 7. 切换上线
└── 8. 监控优化

// 检查MongoDB版本
db.version()

// DocumentDB支持版本：3.6, 4.0, 5.0
// 建议升级到兼容版本

// 检查使用的功能
db.adminCommand({ listCommands: 1 })

// 检查索引
db.collection.getIndexes()

// 检查数据大小
db.stats()

DocumentDB兼容性：
├── 支持的功能
│   ├── CRUD操作
│   ├── 聚合管道
│   ├── 索引
│   ├── 事务（4.0+）
│   └── 变更流
│
└── 不支持/部分支持
    ├── 固定大小集合
    ├── 部分聚合操作符
    ├── MapReduce
    ├── GridFS
    └── 部分地理空间功能

// 评估数据量
db.getCollectionNames().forEach(function(name) {
  const stats = db[name].stats();
  print(name + ": " + stats.size + " bytes, " + stats.count + " documents");
});

// 评估索引
db.getCollectionNames().forEach(function(name) {
  const indexes = db[name].getIndexes();
  print(name + ": " + indexes.length + " indexes");
});

# 创建子网组
aws docdb create-db-subnet-group \
  --db-subnet-group-name my-subnet-group \
  --db-subnet-group-description "Migration subnet group" \
  --subnet-ids subnet-12345 subnet-67890

# 创建集群
aws docdb create-db-cluster \
  --db-cluster-identifier migration-cluster \
  --engine docdb \
  --engine-version 5.0.0 \
  --master-username admin \
  --master-user-password password123 \
  --db-subnet-group-name my-subnet-group \
  --vpc-security-group-ids sg-12345

# 创建实例
aws docdb create-db-instance \
  --db-instance-identifier migration-primary \
  --db-instance-class db.r6g.large \
  --engine docdb \
  --db-cluster-identifier migration-cluster

网络要求：
├── 源数据库可访问
├── 目标DocumentDB可访问
├── 迁移服务器网络连通
├── 安全组规则配置
└── 带宽充足

# 安装mongodump和mongorestore
# Ubuntu/Debian
sudo apt-get install mongodb-tools

# macOS
brew install mongodb-community-tools

# 安装AWS DMS
# 通过AWS控制台或CLI

# 导出数据
mongodump \
  --host source-host \
  --port 27017 \
  --username admin \
  --password \
  --db mydb \
  --out /backup/dump

# 导入数据
mongorestore \
  --host target-cluster.cluster-xxx.docdb.amazonaws.com \
  --port 27017 \
  --username admin \
  --password \
  --ssl \
  --sslCAFile rds-combined-ca-bundle.pem \
  /backup/dump/mydb

# 并行导入（提高速度）
mongorestore \
  --host target-cluster.cluster-xxx.docdb.amazonaws.com \
  --port 27017 \
  --username admin \
  --password \
  --ssl \
  --sslCAFile rds-combined-ca-bundle.pem \
  --numInsertionWorkersPerCollection 8 \
  /backup/dump/mydb

# 导出为JSON
mongoexport \
  --host source-host \
  --port 27017 \
  --username admin \
  --password \
  --db mydb \
  --collection users \
  --out users.json

# 导入JSON
mongoimport \
  --host target-cluster.cluster-xxx.docdb.amazonaws.com \
  --port 27017 \
  --username admin \
  --password \
  --ssl \
  --sslCAFile rds-combined-ca-bundle.pem \
  --db mydb \
  --collection users \
  --file users.json

# 创建复制实例
aws dms create-replication-instance \
  --replication-instance-identifier my-replication-instance \
  --replication-instance-class dms.r5.large \
  --allocated-storage 50

# 创建源端点
aws dms create-endpoint \
  --endpoint-identifier source-mongodb \
  --endpoint-type source \
  --engine-name mongodb \
  --server-name source-host \
  --port 27017 \
  --username admin \
  --password password123

# 创建目标端点
aws dms create-endpoint \
  --endpoint-identifier target-documentdb \
  --endpoint-type target \
  --engine-name docdb \
  --server-name target-cluster.cluster-xxx.docdb.amazonaws.com \
  --port 27017 \
  --username admin \
  --password password123

# 创建迁移任务
aws dms create-replication-task \
  --replication-task-identifier my-migration-task \
  --source-endpoint-arn source-endpoint-arn \
  --target-endpoint-arn target-endpoint-arn \
  --replication-instance-arn instance-arn \
  --migration-type full-load-and-cdc \
  --table-mappings '{
    "rules": [{
      "rule-type": "selection",
      "rule-id": "1",
      "object-locator": {
        "schema-name": "mydb",
        "table-name": "%"
      },
      "rule-action": "include"
    }]
  }'

// 自定义迁移脚本
const { MongoClient } = require('mongodb');

async function migrateCollection(sourceUri, targetUri, dbName, collectionName, batchSize = 1000) {
  const sourceClient = new MongoClient(sourceUri);
  const targetClient = new MongoClient(targetUri, {
    tls: true,
    tlsCAFile: './rds-combined-ca-bundle.pem'
  });
  
  try {
    await sourceClient.connect();
    await targetClient.connect();
    
    const sourceCollection = sourceClient.db(dbName).collection(collectionName);
    const targetCollection = targetClient.db(dbName).collection(collectionName);
    
    let skip = 0;
    let total = 0;
    
    while (true) {
      const docs = await sourceCollection.find({}).skip(skip).limit(batchSize).toArray();
      
      if (docs.length === 0) break;
      
      await targetCollection.insertMany(docs, { ordered: false });
      
      skip += batchSize;
      total += docs.length;
      
      console.log(`Migrated ${total} documents`);
    }
    
    console.log(`Migration completed: ${total} documents`);
    
  } finally {
    await sourceClient.close();
    await targetClient.close();
  }
}

// 使用示例
migrateCollection(
  'mongodb://source-host:27017',
  'mongodb://target-cluster.cluster-xxx.docdb.amazonaws.com:27017',
  'mydb',
  'users'
);

// 导出所有索引
db.getCollectionNames().forEach(function(collectionName) {
  print("Collection: " + collectionName);
  const indexes = db[collectionName].getIndexes();
  indexes.forEach(function(index) {
    printjson(index);
  });
});

// 在目标数据库创建索引
// 根据导出的索引定义创建

// 单字段索引
db.users.createIndex({ email: 1 }, { unique: true });

// 复合索引
db.orders.createIndex({ userId: 1, createdAt: -1 });

// 文本索引
db.articles.createIndex({ content: "text" });

// TTL索引
db.sessions.createIndex(
  { createdAt: 1 },
  { expireAfterSeconds: 3600 }
);

// 原连接字符串
const oldUri = "mongodb://user:pass@source-host:27017/mydb";

// 新连接字符串（DocumentDB）
const newUri = "mongodb://user:pass@target-cluster.cluster-xxx.docdb.amazonaws.com:27017/mydb?ssl=true&replicaSet=rs0";

// 完整配置
const client = new MongoClient(newUri, {
  tls: true,
  tlsCAFile: './rds-combined-ca-bundle.pem',
  replicaSet: 'rs0',
  readPreference: 'secondaryPreferred'
});

// 检查不支持的功能
// 1. 固定大小集合
// MongoDB
db.createCollection("logs", { capped: true, size: 10000000 });

// DocumentDB替代方案
db.createCollection("logs");
// 使用TTL索引实现类似功能
db.logs.createIndex({ createdAt: 1 }, { expireAfterSeconds: 86400 });

// 2. MapReduce
// MongoDB
db.orders.mapReduce(mapFunction, reduceFunction, { out: "results" });

// DocumentDB替代方案
db.orders.aggregate([
  { $group: { _id: "$category", total: { $sum: "$amount" } } },
  { $out: "results" }
]);

驱动版本要求：
├── Node.js: mongodb >= 3.6
├── Python: pymongo >= 3.8
├── Java: mongodb-driver-sync >= 3.12
├── C#: MongoDB.Driver >= 2.9
└── Go: mongo-driver >= 1.3

// 验证文档数量
const sourceCount = await sourceCollection.countDocuments({});
const targetCount = await targetCollection.countDocuments({});
console.log(`Source: ${sourceCount}, Target: ${targetCount}`);

// 验证数据一致性
async function verifyData(sourceCollection, targetCollection) {
  const sourceDocs = await sourceCollection.find({}).sort({ _id: 1 }).toArray();
  const targetDocs = await targetCollection.find({}).sort({ _id: 1 }).toArray();
  
  if (sourceDocs.length !== targetDocs.length) {
    console.error('Document count mismatch');
    return false;
  }
  
  for (let i = 0; i < sourceDocs.length; i++) {
    const source = JSON.stringify(sourceDocs[i]);
    const target = JSON.stringify(targetDocs[i]);
    
    if (source !== target) {
      console.error(`Document mismatch at index ${i}`);
      return false;
    }
  }
  
  console.log('Data verification passed');
  return true;
}

功能测试清单：
├── 连接测试
├── CRUD操作测试
├── 查询功能测试
├── 索引功能测试
├── 事务功能测试
├── 聚合功能测试
└── 应用集成测试

// 性能对比测试
async function performanceTest(collection) {
  const start = Date.now();
  
  for (let i = 0; i < 1000; i++) {
    await collection.findOne({ _id: ObjectId() });
  }
  
  const duration = Date.now() - start;
  console.log(`1000 queries in ${duration}ms`);
}

切换步骤：
├── 1. 停止应用写入源数据库
├── 2. 执行最终数据同步
├── 3. 验证数据一致性
├── 4. 更新应用配置
├── 5. 重启应用服务
├── 6. 验证应用功能
├── 7. 监控系统状态
└── 8. 保留回滚方案

回滚准备：
├── 保留源数据库
├── 记录切换时间点
├── 准备快速回滚脚本
├── 通知相关人员
└── 制定回滚决策标准

迁移后优化：
├── 分析慢查询
├── 优化索引
├── 调整配置参数
├── 监控性能指标
└── 持续优化

成本优化：
├── 选择合适的实例规格
├── 合理配置副本数量
├── 监控资源使用
├── 使用预留实例
└── 优化存储使用

原因和解决：
├── 网络带宽限制：增加带宽
├── 迁移工具配置：优化并行度
├── 目标索引过多：先迁移后建索引
└── 资源不足：增加实例规格

原因和解决：
├── 迁移期间写入：停止写入或使用CDC
├── 网络问题：检查网络稳定性
├── 工具问题：验证迁移工具配置
└── 数据类型问题：检查数据类型兼容性

迁移最佳实践：
├── 充分评估和规划
├── 选择合适的迁移方式
├── 测试验证充分
├── 制定详细的切换计划
├── 保留回滚方案
└── 迁移后持续优化