解析Nodejs中的大型JSON文件

Question

我有一个文件，它以JSON形式存储了许多JavaScript对象，我需要读取该文件，创建每个对象，并对它们做一些事情（在我的例子中，将它们插入数据库）。JavaScript对象可以用一种格式表示：

格式A：

[{name: 'thing1'},
....
{name: 'thing999999999'}]

或格式B：

{name: 'thing1'}         // <== My choice.
...
{name: 'thing999999999'}

请注意， 表示许多JSON对象。我知道我可以将整个文件读入内存，然后使用JSON。parse（）如下所示：

fs.readFile(filePath, 'utf-8', function (err, fileContents) {
  if (err) throw err;
  console.log(JSON.parse(fileContents));
});

然而，文件可能非常大，我更喜欢使用流来完成这一点。我在流中看到的问题是，文件内容在任何时候都可能被分解成数据块，因此如何使用JSON。在这样的对象上解析（）？

理想情况下，每个对象将被读取为一个单独的数据块，但我不确定如何做到这一点。

var importStream = fs.createReadStream(filePath, {flags: 'r', encoding: 'utf-8'});
importStream.on('data', function(chunk) {

    var pleaseBeAJSObject = JSON.parse(chunk);           
    // insert pleaseBeAJSObject in a database
});
importStream.on('end', function(item) {
   console.log("Woot, imported objects into the database!");
});*/

注意，我希望防止将整个文件读取到内存中。时间效率对我来说并不重要。是的，我可以尝试一次读取许多对象并一次插入所有对象，但这是性能调整——我需要一种保证不会导致内存过载的方法，不管文件中包含多少对象。

我可以选择使用FormatA或FormatB或其他，请在回答中详细说明。谢谢




Answer 1

从2014年10月起，您只需执行以下操作（使用JSONStream）——https://www.npmjs.org/package/JSONStream

var fs = require('fs'),
    JSONStream = require('JSONStream'),

var getStream() = function () {
    var jsonData = 'myData.json',
        stream = fs.createReadStream(jsonData, { encoding: 'utf8' }),
        parser = JSONStream.parse('*');
    return stream.pipe(parser);
}

getStream().pipe(MyTransformToDoWhateverProcessingAsNeeded).on('error', function (err) {
    // handle any errors
});

演示：用一个实例来演示：

npm install JSONStream event-stream

data.json：

{
  "greeting": "hello world"
}

你好js:

var fs = require('fs'),
    JSONStream = require('JSONStream'),
    es = require('event-stream');

var getStream = function () {
    var jsonData = 'data.json',
        stream = fs.createReadStream(jsonData, { encoding: 'utf8' }),
        parser = JSONStream.parse('*');
    return stream.pipe(parser);
};

getStream()
    .pipe(es.mapSync(function (data) {
        console.log(data);
    }));

$ node hello.js
// hello world

Answer 2

匿名用户

就像我在想写一个流式JSON解析器会很有趣一样，我也想也许我应该快速搜索一下，看看是否有一个可用的解析器。

原来有。

• JSONStream“streaming JSON.parse and stringify”

因为我刚刚找到它，我显然没有用过它，所以我不能对它的质量发表评论，但我很想知道它是否有效。

它可以考虑下面的JavaScript和<代码>。isString：

stream.pipe(JSONStream.parse('*'))
  .on('data', (d) => {
    console.log(typeof d);
    console.log("isString: " + _.isString(d))
  });

如果流是一个对象数组，这将在对象进入时记录它们。因此，一次只能缓冲一个对象。

Answer 3

要逐行处理文件，您只需要将文件的读取和作用于该输入的代码分离。您可以通过缓冲输入来实现这一点，直到您碰到换行符。假设我们每行有一个JSON对象（基本上是格式B）：

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

stream.on('data', function(d) {
    buf += d.toString(); // when data is read, stash it in a string buffer
    pump(); // then process the buffer
});

function pump() {
    var pos;

    while ((pos = buf.indexOf('\n')) >= 0) { // keep going while there's a newline somewhere in the buffer
        if (pos == 0) { // if there's more than one newline in a row, the buffer will now start with a newline
            buf = buf.slice(1); // discard it
            continue; // so that the next iteration will start with data
        }
        processLine(buf.slice(0,pos)); // hand off the line
        buf = buf.slice(pos+1); // and slice the processed data off the buffer
    }
}

function processLine(line) { // here's where we do something with a line

    if (line[line.length-1] == '\r') line=line.substr(0,line.length-1); // discard CR (0x0D)

    if (line.length > 0) { // ignore empty lines
        var obj = JSON.parse(line); // parse the JSON
        console.log(obj); // do something with the data here!
    }
}

每次文件流从文件系统接收数据时，它都被存储在一个缓冲区中，然后调用pump。

如果缓冲区中没有换行符，pump只返回，不做任何操作。下次流获取数据时，更多的数据（可能还有一个换行符）将被添加到缓冲区，然后我们将拥有一个完整的对象。

如果有换行符，泵将缓冲区从开头切到换行符，并将其交给处理。然后再次检查缓冲区中是否有另一个换行符（while循环）。通过这种方式，我们可以处理当前块中读取的所有行。

最后，每个输入行调用一次process。如果存在，它会去掉回车符（以避免行尾问题–LF vs CRLF），然后调用JSON。解析一行。在这一点上，你可以对你的对象做任何你需要做的事情。

请注意，JSON。parse对它接受的输入是严格的；必须用双引号引用标识符和字符串值。换句话说，{name:'thing1'}将抛出一个错误；您必须使用{“name”：“thing1”}。

因为一次内存中的数据不会超过一块，所以这将是非常高效的内存。它也会非常快。快速测试显示我在15毫秒内处理了10000行。