如何在Node.js中为一个非常大(> 1GB)的文件的每一行运行异步函数

Question

假设您有一个巨大的(> 1GB)CSV记录ID:

655453
4930285
493029
4930301
493031
...

对于每个人,id您希望进行REST API调用以获取记录数据,在本地转换它,并将其插入本地数据库.

你如何使用Node.js' 可读Stream？

我的问题基本上是这样的:你如何逐行读取一个非常大的文件,为每一行运行一个异步函数,并且[可选]能够从特定的行开始读取文件？

从以下Quora问题我开始学习使用fs.createReadStream:

http://www.quora.com/What-is-the-best-way-to-read-a-file-line-by-line-in-node-js

var fs = require('fs');
var lazy = require('lazy');

var stream = fs.createReadStream(path, {
  flags: 'r',
  encoding: 'utf-8'
});

new lazy(stream).lines.forEach(function(line) {
  var id = line.toString();
  // pause stream
  stream.pause();
  // make async API call...
  makeAPICall(id, function() {
    // then resume to process next id
    stream.resume();
  });
});

但是,该伪代码不起作用,因为该lazy模块强制您读取整个文件(作为流,但没有暂停).所以这种方法似乎不会起作用.

另一件事是,我希望能够从特定的行开始处理这个文件.原因是,处理每个id(进行api调用,清理数据等)每个记录可能需要半秒钟,因此我不希望每次都从文件的开头开始.我正在考虑使用的天真方法是捕获最后处理的id的行号,并保存它.然后,当您再次解析文件时,您将逐行流式传输所有ID,直到找到您中断的行号,然后您开展makeAPICall业务.另一种天真的方法是编写小文件(比如100个ID)并一次处理一个文件(足够小的数据集可以在没有IO流的情况下在内存中完成所有操作).有一个更好的方法吗？

我可以看到这是如何变得棘手(以及node-lazy进来的地方),因为chunkin stream.on('data', function(chunk) {});可能只包含一行的一部分(如果bufferSize很小,每个chunk可能是10行但是因为它id是可变长度,它可能只是9.5行或其他).这就是为什么我想知道上述问题的最佳方法是什么.

Answer 1

我想你不需要使用node-lazy. 这是我在Node 文档中找到的内容：

事件：data

function (data) { }

该data事件发出一个Buffer（默认情况下）或一个stringif setEncoding()被使用。

这意味着您调用setEncoding()流，然后您的data事件回调将接受字符串参数。然后在这个回调中你可以调用 use.pause()和.resume()methods。

伪代码应如下所示：

stream.setEncoding('utf8');
stream.addListener('data', function (line) {
    // pause stream
    stream.pause();
    // make async API call...
    makeAPICall(line, function() {
        // then resume to process next line
        stream.resume();
    });
})

尽管文档没有明确指定流是逐行读取的，但我认为文件流就是这种情况。至少在其他语言和平台中，文本流是这样工作的，我认为 Node 流没有理由有所不同。