如何在MongoDB中执行数据清理并归档历史数据？

以下脚本基于MongoDB环境编写，默认文档包含用于判断创建时间的create_time字段，可根据实际业务字段名替换。

前置注意事项 #

• 执行脚本前务必对transactions集合做全量备份，避免误操作导致数据丢失
• 大数量级场景建议在业务低峰期执行，避免占用过多数据库资源影响正常业务
• 如需保留索引，可提前为transactions_archive集合创建和原集合一致的索引规则

单次执行完整脚本（适合数据量小于10万的场景） #

// 计算6个月前的时间节点
const sixMonthsAgo = new Date();
sixMonthsAgo.setMonth(sixMonthsAgo.getMonth() - 6);

// 第一步：将符合条件的旧数据归档到transactions_archive集合，已有归档数据不覆盖
db.transactions.aggregate([
  { $match: { create_time: { $lt: sixMonthsAgo } } },
  { 
    $merge: {
      into: "transactions_archive",
      whenMatched: "keepExisting",
      whenNotMatched: "insert"
    }
  }
])

// 第二步：校验归档数据和待删除数据数量一致后执行删除
const toDeleteCount = db.transactions.countDocuments({ create_time: { $lt: sixMonthsAgo } });
const archivedCount = db.transactions_archive.countDocuments({ create_time: { $lt: sixMonthsAgo } });

if (toDeleteCount === archivedCount) {
  db.transactions.deleteMany({ create_time: { $lt: sixMonthsAgo } });
  print(`执行成功，共归档并清理 ${toDeleteCount} 条旧数据`);
} else {
  print(`数据校验不通过，待删除 ${toDeleteCount} 条，实际归档 ${archivedCount} 条，已终止删除操作`);
}

分批执行脚本（适合数据量大于10万的大集合场景） #

// 计算6个月前的时间节点
const sixMonthsAgo = new Date();
sixMonthsAgo.setMonth(sixMonthsAgo.getMonth() - 6);

// 每批次处理条数，可根据数据库性能调整
const batchSize = 1000;
let totalProcessed = 0;

while (true) {
  // 读取当前批次需要处理的文档ID列表
  const batchIds = db.transactions.find(
    { create_time: { $lt: sixMonthsAgo } },
    { _id: 1 }
  ).limit(batchSize).toArray().map(item => item._id);

  if (batchIds.length === 0) {
    break;
  }

  // 归档当前批次数据
  db.transactions.aggregate([
    { $match: { _id: { $in: batchIds } } },
    { 
      $merge: {
        into: "transactions_archive",
        whenMatched: "keepExisting",
        whenNotMatched: "insert"
      }
    }
  ]);

  // 删除当前批次的原集合数据
  const deleteRes = db.transactions.deleteMany({ _id: { $in: batchIds } });
  totalProcessed += deleteRes.deletedCount;

  // 每批次执行完成后暂停1秒，降低数据库压力
  sleep(1000);
}

print(`分批执行完成，共归档并清理 ${totalProcessed} 条旧数据`);

内容的提问来源于stack exchange，提问作者Pradip Kumar