节点Js的 MongoDB 批量数据迁移

Mongodb计划迁移工具.

关于

MongoBulkDataMigration 是一行的MongoDb迁移工具 用于您一发的迁移. 它速度快,可恢复,自动处理回滚.

为什么是蒙哥堡?

• 🚀 快点: 建造在 蒙戈散货业务 少强调蒙戈插座。 Bulk在编组操作时适应了大规模更新.

• 🏔️ 可缩放:处理100万/10亿文件不是问题. 备份在一个专用的集合中完成. Bulks可以被挤出.

• 🔒 安全而轻松: 备份和还原工作逐批逐步完成. 凡取出并投出的东西,都是作为备份保存的. 在MBDM上所构建的脚本是明确而有重点的.

• 🔄 可恢复: 如果一个脚本崩溃, 可以恢复它 (断开)。 。 。 . 从设计上来说,在更新或回滚模式中运行两次是安全的.

• 💝 最小:一个"统一"的扩展脚本由平台处理来写出更少的代码. 还有,测试工具 doRollbackAndAssertForInitialState 将简化 Jest 和 Chai 框架的测试写法.

• 🏔️ 总体准备情况: 您可以使用简单的查询或聚合管道获取

使用 MBDM 的前提条件

MBDM满足大多数需要,但不是全部需要. 先决条件:

• MBDM 无法插入新文档 (您只能更新或删除文档)
• MBDM 只能更新 一次收藏1个
• MBDM 只能更新文档 一次 同一移民身份证 (除非您手动清理备份)
  • 查询最好不能匹配已经更新的文档 (用于恢复)

✅ MBDM 能力

最大特性使 MBDM 强大 :

• MBDM 能够 自动回滚 至完全相同的状态预测值:
  • 页:1 $set 业务
  • 页:1 $unset 业务
  • 用于文档删除操作
  • ...对于任何更高级的,你仍然可以使用一个习惯回滚 rollback 函数
• MBDM 接受基本的蒙哥查询,或 管道合计, (中文(简体) ),
• MBDM 可以在一个 自动调用,带有定义 maxConcurrentUpdateCalls,允许您调用一个复杂的同步代码.
• MBDM 让弦乐脚本快速写作 QQ

支持日志

MBDM 数据管理

开始

安装 MBDM

使用 npm

npm install --save-dev @360-l/mongo-bulk-data-migration

运行迁移 update() 财务报告和审定财务报表 rollback()

MBDM 期望连接到您的数据库、 ID( 字符串) 以及目标收藏来处理 .

import { MongoBulkDataMigration } from "@360-l/mongo-bulk-data-migration";

// connect to db
// handle script input

const migration = new MongoBulkDataMigration({
    db: mongoClient,       // MongoClient established instance
    collection: "myCol",   // Collection where there will be an update
    id: "uid_migration",   // Required to rollback storage
    ... // see below for query/update
});

// Do the update
await migration.update();

// Or revert updated properties (only) in all updated documents
await migration.rollback();

MBDM不提供CLI工具. 你可以考虑创建一些结构抽象代码来运行它像. 例如:

node ./myscript.ts --action [update|rollback]

简单 $set 示例

这个迁移将设定 { total: 0 } 每个医生没有0.

注:回滚将自动取消回放总计.

new MongoBulkDataMigration<Score>({
  db,
  id: 'scores_set_total',
  collectionName: 'scores',
  projection: {},
  query: { total: { $exists: false } },
  update: { $set: { total: 0 } },
});

QQ 选项

new MongoBulkDataMigration({ ..., options: { ... } })

• maxBulkSize (默认: 1000): 要执行的批量更新大小 。 1000是一个很好的数字。 如果您的更新量很大, 请考虑减少, 以避免运行高内存 。 如果文档是微小的,10克可能会在巨大的数据库中稍作改进.
• dontCount (默认: 虚假): 将跳过初始过滤器 count() 或合计 $count。如果禁用,日志不会打印全部进度 .
• projectionBackupFilter (默认: 无): 要保存和自动回滚的过滤属性 。 如果您的更新需要虚拟字段, 则有必要 .
• bypassRollbackValidation 财务报告和审定财务报表 bypassUpdateValidation (默认: 假): 将验证等级设置为"关闭"再设置为"中度".
• throttle (默认: 0):在批量更新之间睡眠的时间以ms计. 用这来减轻数据库的压力.
• continueOnBulkWriteError (default: false):会以批量的误差来继续迁移.
• batchScanSize (默认: 无): 连续扫描收藏 _id 范围 batchScanSize 文档(例如:. 100_000)而不是一个单一的查询. 可用于长期迁移,避免套接字超时.
• hint (默认: 无): 用于迁移查询及其计数的强制索引

高级用途

MBDM支持Async更新查询,自定义回滚,聚合管道...

简单的$set, 有更新调用和投影

这一迁移将汇总2个预测领域 scoreA 财务报告和审定财务报表 scoreB 用于所有文档(无过滤器).

注意: 回滚会自动回落 scoreA 财务报告和审定财务报表 scoreB 并重设 total。 。 。 .

new MongoBulkDataMigration<Score>({
  db,
  id: 'scores_total_new_field',
  collectionName: 'scores',
  projection: { scoreA: 1, scoreB: 1 },
  query: FETCH_ALL,
  update: (doc) => ({
    $set: {
      total: doc.scoreA + doc.scoreB,
    },
  }),
});

文档的有条件更新

这种迁移将根据外部条件更新文件. 还给我 NO_UPDATE 符号以宣布不需要更新.

import { MongoBulkDataMigration, NO_UPDATE } from "@360-l/mongo-bulk-data-migration";

new MongoBulkDataMigration<Score>({
  db,
  id: 'conditional_update',
  collectionName: 'scores',
  query: FETCH_ALL,
  update: async (doc) => {
    if (! await shouldUpdate(doc)) { return NO_UPDATE; }
    return { $set: { satus: "updated" } };
  },
});

async function shouldUpdate(...) // external check

自动恢复( E)query: FETCH_ALL(中文(简体) )

如果您无法提供仅获取非迁移文档的查询( 如 : newField:{$exists:false},您可以让 MBDM 继续按照上一个 _id 已处理:

import { MongoBulkDataMigration, FETCH_ALL } from "@360-l/mongo-bulk-data-migration";
...
new MongoBulkDataMigration<Score>({
    db,
    id: "update_all_totals",
    collectionName: "scores",
    projection: { total:1 },
    // Automatic resume, iso {}
    query: FETCH_ALL,
    update: () => ({ total: total + 1 }),
});

删除文档( E)update: DELETE_OPERATION(中文(简体) )

此迁移将删除负数的 doc total。 。 。 .

注意:回滚会自动恢复完整的文档.

import { MongoBulkDataMigration, DELETE_OPERATION } from "@360-l/mongo-bulk-data-migration";
...
new MongoBulkDataMigration<Score>({
    db,
    id: "delete_negative_total",
    collectionName: "scores",
    projection: {}, // Everything needs to be projected
    query: { total: { $lt: 0 } },
    update: DELETE_OPERATION,
});

利用汇总管道(query(中文(简体) )

query 期望查询(对象)或聚合管道(阵列). 说明: projection 什么都不会做.

示例:更新 totalGames 对应表格

import { MongoBulkDataMigration } from "@360-l/mongo-bulk-data-migration";
...
new MongoBulkDataMigration<Score>({
    db,
    id: "delete_negative_total",
    collectionName: "scores",
    projection: { "games.value": 1, totalGames: 1 },
    query: [
        { $lookup: { as: "games", ... } },
        { $match: { "games.value": "xxx" } },
    ],
    update: (doc) => ({
        $set: {
            totalGames: doc.games.value
        }
    }),
    options: {
        // Necessary to save only `totalGames`, and not auto-restore uncexisting `games` field
        projectionBackupFilter: ["totalGames"]
    }
});

删除收藏( E)operation: DELETE_COLLECTION(中文(简体) )

收藏会更名为备份收藏. 当你回滚时, 收藏将简单地重新命名为回滚

import { MongoBulkDataMigration, DELETE_COLLECTION } from "@360-l/mongo-bulk-data-migration";
...
const migration status = new MongoBulkDataMigration<Score>({
    db,
    id: "delete_collection_scores",
    collectionName: "scores",
    operation: DELETE_COLLECTION,
});
console.log(status); // { ok: 1 }

* 简易测试

MBDM提供简化测试文件. 只要写你期望的, 用户会确保 数据库恢复到initia状态. 它支持Jest和Chai测试库(expect) (中文(简体) ).

注:对分类文件的预期是按 _id

it('should not modify pages with existing translations', async () => {
  const docs = [{ _id: new ObjectId(), value: 'invalid' }];
  await collection.insertMultiple(docs);

  await dataMigration.update();

  // Test what you expect
  const updatedDocs = await collection.find({}).toArray();
  expect(updatedDocs).toEqual([{ _id: new ObjectId(), value: 'valid' }]);

  // Test rollback will work
  await doRollbackAndAssertForInitialState(dataMigration, docs, { expect });
});