节点 CSV 拉解析器

Question

我需要从 Node.JS 解析一个 CSV 文档，为每条记录（= 每一行）执行数据库操作。但是，我无法使用拉取方法找到合适的 CSV 解析器，或者至少是在解析下一行之前等待我的记录操作的推送方法。

我看过csv-parse、csvtojson、csv-streamify，但它们似乎都在没有任何流控制的情况下以连续流的形式推送事件。如果解析一个 1000 行的 CSV 文档，我基本上可以快速获得所有 1000 个回调。对于每条记录，我执行一个返回承诺的操作。目前，我不得不求助于将我所有的承诺推入一个数组，并且在获得done/end 事件后，我还等待Promise.all(myOperations) 知道文档何时已被完全处理。但这不是很好，而且，我更喜欢在获取下一条记录之前一次解析一行并完全处理它，而不是同时处理所有记录 - 它很难调试并且使用大量内存相反简单地按顺序处理每条记录。

那么，有没有支持拉模式的 CSV 解析器，或者是获取任何基于流的 CSV 解析器的方法（最好是 csvtojson，因为这是我目前正在使用的）只为新记录生成事件我的前一条记录的处理程序何时完成（使用承诺）？

Answer 1

我自己解决了这个问题，方法是创建自己的 Writable 并将 CSV 解析器传送给它。我的 write 方法完成了它的工作，并为传递给 _write() 的节点回调包装了一个承诺（这里使用 Q.nodeify 实现）：

class CsvConsumer extends stream.Writable {
    _write(data, encoding, cb) {
        console.log('Got data: ', data);

        Q.delay(1000).then(() => {
            console.log('Waited 1 s');
        }).nodeify(cb);
    }
}

csvtojson()
    .fromStream(is)
    .pipe(new CsvConsumer())
    .on('finish', err => {
        if (err) {
            console.log('Error!');
        } else {
            console.log('Done!');
        }
    });

这将逐行处理：

Got data: {"a": "1"}
Waited 1 s
Got data: {"a": "2"}
Waited 1 s
Got data: {"a": "3"}
Waited 1 s
Done!

Answer 2

如果您想异步处理每一行，您可以使用节点的原生 LineReader 来实现。

const lineStream = readline.createInterface({
  input: fs.createReadStream('data/test.csv'),
});

lineStream.on('line', (eachLine) =>{
    //process each line
});

如果你想以同步方式做同样的事情，你可以使用line-by-line。它不会将整个文件缓冲到内存中。它提供事件处理程序来暂停和恢复“line”发出事件。

lr.on('line', function (line) {
    // pause emitting of lines...
    lr.pause();

    // ...do your asynchronous line processing..
    setTimeout(function () {

        // ...and continue emitting lines. (1 sec delay)
        lr.resume();
    }, 1000);
});