Node.js 自动化数据迁移工具-高效同步 MySQL 与 MongoDB 数据

data-migrator/
├── config/
│   └── mapping.yaml         # 字段映射配置
├── .env                     # 环境变量配置
├── migrate.js               # 主迁移脚本
├── mysql.js                 # MySQL 查询工具
├── mongo.js                 # MongoDB 插入工具
└── utils.js                 # 工具函数

MYSQL_HOST=localhost
MYSQL_USER=root
MYSQL_PASSWORD=123456
MYSQL_DATABASE=testdb

MONGO_URI=mongodb://localhost:27017/testdb

users:
  collection: users
  fields:
    id: _id
    username: name
    email: email
    created_at: createdAt
orders:
  collection: orders
  fields:
    id: _id
    user_id: userId
    amount: totalAmount
    created_at: createdAt

const mysql = require('mysql2/promise');
require('dotenv').config();

const pool = mysql.createPool({
  host: process.env.MYSQL_HOST,
  user: process.env.MYSQL_USER,
  password: process.env.MYSQL_PASSWORD,
  database: process.env.MYSQL_DATABASE,
});

async function fetchTableData(table) {
  const [rows] = await pool.query(`SELECT * FROM \`${table}\``);
  return rows;
}

module.exports = { fetchTableData };

const mongoose = require('mongoose');
require('dotenv').config();

mongoose.connect(process.env.MONGO_URI, {
  useNewUrlParser: true,
  useUnifiedTopology: true,
});

async function insertMany(collectionName, docs) {
  const Model = mongoose.model(collectionName, new mongoose.Schema({}, { strict: false }), collectionName);
  await Model.insertMany(docs);
}

module.exports = { insertMany };

function transformRow(row, fieldMap) {
  const newRow = {};
  for (const [source, target] of Object.entries(fieldMap)) {
    newRow[target] = row[source];
  }
  return newRow;
}

module.exports = { transformRow };

const fs = require('fs');
const yaml = require('js-yaml');
const { fetchTableData } = require('./mysql');
const { insertMany } = require('./mongo');
const { transformRow } = require('./utils');

// 加载字段映射配置
const mappings = yaml.load(fs.readFileSync('./config/mapping.yaml', 'utf8'));

async function migrate() {
  for (const [table, config] of Object.entries(mappings)) {
    console.log(`📥 开始迁移表: ${table}`);

    const rows = await fetchTableData(table);
    const transformed = rows.map(row => transformRow(row, config.fields));

    await insertMany(config.collection, transformed);
    console.log(`✅ 已迁移 ${rows.length} 条记录到 ${config.collection}`);
  }

  console.log('🎉 所有迁移任务完成！');
  process.exit(0);
}

migrate().catch(err => {
  console.error('❌ 迁移失败:', err);
  process.exit(1);
});