如何使用 Sequelize 和 node.js 进行批量插入

Posted 2023-03-07

【中文标题】如何使用 Sequelize 和 node.js 进行批量插入

【英文标题】：How to do Bulk insert using Sequelize and node.js

【发布时间】：2015-06-10 07:08:59

【问题描述】：

js + sequelize 使用 JSON 插入 280K 行数据。 JSON 是一个 280K 的数组。有没有办法批量插入块。我看到更新数据需要很长时间。当我试图将数据减少到 40K 行时，它运行得很快。我是否采取了正确的方法。请指教。我使用 postgresql 作为后端。

PNs.bulkCreate(JSON_Small)
        .catch(function(err) 
            console.log('Error ' + err);
        )
        .finally(function(err) 
            console.log('FINISHED  + ' \n +++++++ \n');

        );

【问题讨论】：

这里有同样的问题，有答案：***.com/questions/33129677/…

【参考方案1】：

您可以使用 Sequelize 内置的bulkCreate 方法来实现这一点。

User.bulkCreate([
   username: 'barfooz', isAdmin: true ,
   username: 'foo', isAdmin: true ,
   username: 'bar', isAdmin: false 
]).then(() =>  // Notice: There are no arguments here, as of right now you'll have to...
  return User.findAll();
).then(users => 
  console.log(users) // ... in order to get the array of user objects
)

Sequelize | Bulk Create and Update

【参考方案2】：

我利用异步库的cargo 实用程序一次加载多达 1000 行。请参阅以下代码将 csv 加载到数据库中：

var fs = require('fs'),
    async = require('async'),
    csv = require('csv');

var input = fs.createReadStream(filename);
var parser = csv.parse(
  columns: true,
  relax: true
);
var inserter = async.cargo(function(tasks, inserterCallback) 
    model.bulkCreate(tasks).then(function() 
        inserterCallback(); 
      
    );
  ,
  1000
);
parser.on('readable', function () 
  while(line = parser.read()) 
    inserter.push(line);
  
);
parser.on('end', function (count) 
  inserter.drain = function() 
    doneLoadingCallback();
  
);
input.pipe(parser);

【参考方案3】：

如果您真的想使用 bulkInsert，那么我之前的回答就足够了。但是，如果您有大量数据，您将耗尽内存！最好使用一些内置的数据库方法是最好的。问题是您将所有数据加载到内存中，直到 bulkCreate 执行。如果你有一百万行，你可能会在它执行之前耗尽内存。即使如此，如果您使用类似async.cargo 的方式将其排队，您仍将等待数据库回复您，同时数据异步消耗您的所有内存。

我的解决方案是放弃 sequelize 以加载数据（至少在他们实现流媒体或其他东西之前（请参阅他们的 github issue #2454））。我最终创建了db-streamer，但它现在只有 pg 支持。你会想看看streamsql 的mysql。

【参考方案4】：

以下问题的答案与您在此处需要的答案相同： NodeJS, promises, streams - processing large CSV files

使用流读取数据并对其进行解析； 使用spex中的stream.read和sequence方法的组合来读取流并一一执行查询。