使用 nodejs 脚本的 SQL 到 MongoDB 迁移

Question

我是 nodejs 的新手，我目前正在执行 sql 到 mongodb 的迁移。我创建了一个脚本来从 sql 查询将数据加载到 mongodb。我使用来自 Google 的示例代码创建了脚本，并且它正在运行。但我面临以下问题，需要解决此问题。

我有一个 sql 查询数组，如果任何查询有任何语法问题或查询结果中有任何错误，我不需要运行这些查询。 （假设第二个查询有语法问题，则无需将第一个查询的数据加载到 mongo，目前它在我的情况下加载）。基本上，如果任何查询有任何问题，则无需将结果加载到 mongo 集合中。如果 mongo 方面有任何问题，也不需要提交事务。

如果有任何错误，我在这里使用了 mongo 事务来回滚数据。请找到以下代码，我们将不胜感激。sql 和 mongo 凭据仅是模拟数据。

配置文件代码

  var mongoCollection = 'collectionName';
  exports.mongoCollection = mongoCollection;

var queryList = [
'sample query one',
'sample query two ' 
];
exports.queryList = queryList;

主脚本代码

  var MongoClient = require('mongodb').MongoClient;
  var sql = require('mysql');

  const config = require('./assets/config');

var sqlConfig = {
user: 'username',
password: 'password',
server: 'servername',
database: 'databasename',
port: 'portname',
multipleStatements: true
};

async function transaction() {
  const mongodbUrl = 'mongourl';
   const client = await MongoClient.connect(mongodbUrl, {useNewUrlParser: true}, {useUnifiedTopology: 
true});
 const db = client.db();

config.queryList.forEach(query => {
    new sql.ConnectionPool(sqlConfig).connect().then(pool => {
        return pool.request().query(query)
    }).then(result => {
        (async()=>{
            const session = client.startSession();
            session.startSession({
                readConcers: {level: 'snapshot'},
                writeConcern: {w: 'majority'}
            });
            try {
                const collection = client.db('mongodbName').collection(config.mongoCollection);
                await collection.insertMany(result.recordset, {session});
                await session.commitTransaction();
                session.emdSession();
                console.log('transaction completed');
            }catch(error){
                await session.abortTransaction();
                session.endSession();
                console.log('transaction aborted');
                throw error;
            }
        });
        sql.close();
    }).catch(error => {
        sql.close();
        throw error;
    })
});
};

transaction();

Answer 1

根据数据量，您可能会考虑将流程分为两部分

• 从mySql中获取数据
• 如果没有错误，加载到 Mongo

这样你就不必回滚 mongo 写入

您也可以利用default mongo pool size (5) 并在mySQL 端使用池。 目前，此代码正在为每个选择创建一个池，这不是最佳的

config.queryList.forEach(query => {
    new sql.ConnectionPool(sqlConfig).connect().then(pool => {//<-New pool per query?
        return pool.request().query(query)
    })
})

相反，您可以设置一次池，per the mySql documentation

看起来该驱动程序只有一个回调 api，但您可以承诺查询以使其更易于使用。

所以把它们放在一起，你可以尝试这样的事情（这不是工作/测试过的代码，只是一个建议）

var MongoClient = require('mongodb').MongoClient;
var sql = require('mysql');

const config = require('./assets/config');

var pool  = sql.createPool({
  connectionLimit : 5,
  host            : 'servername',
  user            : 'username',
  password        : 'password',
  database        : 'databasename'
});

async function transaction() {
  try{
    const mongodbUrl = 'mongourl';
    const client = await MongoClient.connect(mongodbUrl, {useNewUrlParser: true}, {useUnifiedTopology: true});
    const db = client.db();
    const collection = client.db('mongodbName').collection(config.mongoCollection);

    //Map your query list to an array of runSql promises
    //this will complete when all queries return, and jump to the catch if any fail
    let results = await Promise.all(config.queryList.map(runSql))
    
    //Map the results to an array of mongo inserts
    let inserts = await Promise.all(results.map(r=>collection.insertMany(r.recordset)))
    
    //Close all connections
    pool.end((err)=>err?console.err(err):console.log('MySQL Closed'))
    client.close((err)=>err?console.err(err):console.log('MongoDB Closed'))
  }
  catch(err){
    console.error(err)
  }
};

transaction();

function runSql(queryStr){
    return new Promise((resolve, reject)=>{
        pool.query(queryStr, function (error, results, fields){
            error?reject(error):resolve(results)
        })  
    })
}

如果数据量是一个问题，您可能需要查看getting streams from your mySql selects 而不仅仅是运行它们