读取流中的csv文件并将它们存储在数据库中答案

【问题标题】：Read csv files in stream and store them in database读取流中的csv文件并将它们存储在数据库中
【发布时间】：2015-09-10 11:08:55
【问题描述】：

我有一些巨大的 csv 文件，我需要将它们存储在 mongo 数据库中。因为这些文件太大，我需要使用stream。我在数据写入数据库时暂停了流。

var fs = require('fs');
var csv = require('csv');
var mongo = require('mongodb');

var db = mongo.MongoClient.connect...

var readStream = fs.createReadStream('hugefile.csv');
readStream.on('data', function(data) {
  readStream.pause();
  csv.parse(data.toString(), { delimiter: ','}, function(err, output) {
    db.collection(coll).insert(data, function(err) {
      readStream.resume();
    });
  });
});
readStream.on('end', function() {
  logger.info('file stored');
});

但是csv.parse 抛出一个错误，因为我需要逐行读取文件以将它们作为 csv 处理，并转换为 json 用于 mongodb。也许我不应该暂停它们，而是使用界面。我还没有找到任何解决方案。

任何帮助将不胜感激！

【问题讨论】：

标签： node.js csv stream

【解决方案1】：

我认为您可能希望从原始数据流中创建行流。

这是拆分包中的一个示例。 https://www.npmjs.com/package/split

fs.createReadStream(file)
.pipe(split())
.on('data', function (line) {
  //each chunk now is a seperate line! 
})

适应你的例子，它可能看起来像这样

var readStream = fs.createReadStream('hugefile.csv');
var lineStream = readStream.pipe(split());
lineStream.on('data', function(data) {
    //remaining code unmodified

【讨论】：

【解决方案2】：

我不确定 bulk() 是否早在 15 年就已经存在，但是任何尝试从大量来源导入项目的人都应该考虑使用它们。

var fs = require('fs');
var csv = require('fast-csv');
var mongoose = require('mongoose');

var db = mongoose.connect...

var counter = 0;        // to keep count of values in the bulk()
const BULK_SIZE = 1000;
var bulkItem = Item.collection.initializeUnorderedBulkOp();

var readStream = fs.createReadStream('hugefile.csv');
const csvStream = csv.fromStream(readStream, { headers: true });
csvStream.on('data', data => {
    counter++;
    bulkOrder.insert(order);

    if (counter === BATCH_SIZE) {
      csvStream.pause();
      bulkOrder.execute((err, result) => {
        if (err) console.log(err);
        counter = 0;
        bulkItem = Item.collection.initializeUnorderedBulkOp();
        csvStream.resume();
      });
    }
  }
});

【讨论】：