解析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.parse()在这些对象上使用？

理想情况下,每个对象都将被视为一个单独的数据块,但我不知道如何做到这一点.

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

要逐行处理文件,您只需要解除文件的读取和作用于该输入的代码.您可以通过缓冲输入直到您换到换行符来完成此操作.假设我们每行有一个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则只返回而不做任何操作.下次流获取数据时,将向缓冲区添加更多数据(可能还有换行符),然后我们将拥有一个完整的对象.

如果有换行符,pump请将缓冲区从开头切换到换行符并将其移交给换行符process.然后它再次检查缓冲区中是否有另一个换行符(while循环).通过这种方式,我们可以处理当前块中读取的所有行.

最后,process每个输入行调用一次.如果存在,它将去除回车符(为避免出现行结尾问题 - LF与CRLF),然后调用JSON.parse一行.此时,您可以对对象执行任何操作.

请注意,JSON.parse它接受的输入是严格的; 您必须使用双引号引用标识符和字符串值.换句话说,{name:'thing1'}会抛出一个错误; 你必须使用{"name":"thing1"}.

因为一次只有一块数据存在于内存中,所以这将非常节省内存.它也会非常快.快速测试显示我在15ms内处理了10,000行.

Answer 2

正如我认为编写流式JSON解析器会很有趣,我也想我也许应该快速搜索一下是否已有可用的解析器.

原来有.

• JSONStream "流JSON.parse和stringify"

因为我刚发现它,我显然没有使用它,所以我不能评论它的质量,但我会有兴趣听听它是否有效.

它确实可以考虑以下CoffeeScript:

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

如果流是对象数组,这将在对象进入时记录对象.因此,唯一缓冲的是一次一个对象.

Answer 3

截至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"
}

hello.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 4

我有类似的要求,我需要读取节点js中的大型json文件并以块的形式处理数据并调用api并保存在mongodb中.inputFile.json就像:

{
 "customers":[
       { /*customer data*/},
       { /*customer data*/},
       { /*customer data*/}....
      ]
}

现在我使用JsonStream和EventStream同步实现这一点.

var JSONStream = require("JSONStream");
var es = require("event-stream");

fileStream = fs.createReadStream(filePath, { encoding: "utf8" });
fileStream.pipe(JSONStream.parse("customers.*")).pipe(
  es.through(function(data) {
    console.log("printing one customer object read from file ::");
    console.log(data);
    this.pause();
    processOneCustomer(data, this);
    return data;
  }),
  function end() {
    console.log("stream reading ended");
    this.emit("end");
  }
);

function processOneCustomer(data, es) {
  DataModel.save(function(err, dataModel) {
    es.resume();
  });
}

Answer 5

我意识到你想尽可能避免将整个JSON文件读入内存,但是如果你有可用的内存,那么在性能方面可能不是一个坏主意.在json文件上使用node.js的require()非常快速地将数据加载到内存中.

我运行了两个测试,以查看从81MB geojson文件中打印每个功能的属性时的性能.

在第一次测试中,我使用了将整个geojson文件读入内存var data = require('./geo.json').这需要3330毫秒,然后从每个功能打印出一个属性需要804毫秒,总计4134毫秒.但是,似乎node.js使用了411MB的内存.

在第二次测试中,我使用@ arcseldon的答案与JSONStream +事件流.我修改了JSONPath查询以仅选择我需要的内容.这次内存永远不会高于82MB,但是现在整个过程需要70秒才能完成!

Answer 6

如果您可以控制输入文件，并且它是一个对象数组，那么您可以更轻松地解决这个问题。安排输出文件，每条记录占一行，如下所示：

[
   {"key": value},
   {"key": value},
   ...

这仍然是有效的 JSON。

然后，使用 node.js readline 模块一次处理一行。

var fs = require("fs");

var lineReader = require('readline').createInterface({
    input: fs.createReadStream("input.txt")
});

lineReader.on('line', function (line) {
    line = line.trim();

    if (line.charAt(line.length-1) === ',') {
        line = line.substr(0, line.length-1);
    }

    if (line.charAt(0) === '{') {
        processRecord(JSON.parse(line));
    }
});

function processRecord(record) {
    // Process the records one at a time here! 
}

Answer 7

我编写了一个可以做到这一点的模块，称为BFJ。具体来说，该方法bfj.match可用于将大型流分解为离散的JSON块：

const bfj = require('bfj');
const fs = require('fs');

const stream = fs.createReadStream(filePath);

bfj.match(stream, (key, value, depth) => depth === 0, { ndjson: true })
  .on('data', object => {
    // do whatever you need to do with object
  })
  .on('dataError', error => {
    // a syntax error was found in the JSON
  })
  .on('error', error => {
    // some kind of operational error occurred
  })
  .on('end', error => {
    // finished processing the stream
  });

在这里，bfj.match返回一个可读的对象模式流，该流将接收已解析的数据项，并传递3个参数：

1. 包含输入JSON的可读流。

2. 谓词，指示已解析JSON中的哪些项目将被推送到结果流。

3. 一个options对象，指示输入为换行符分隔的JSON（这是为了处理问题中的格式B，格式A不是必需的）。

调用后，bfj.match将首先从输入流中解析JSON，然后使用每个值调用谓词，以确定是否将该项目推送到结果流。谓词传递了三个参数：

1. 属性键或数组索引（这将undefined用于顶级项）。

2. 价值本身。

3. JSON结构中项目的深度（顶级项目为零）。

当然，根据需要，也可以根据需要使用更复杂的谓词。如果要对属性键执行简单匹配，还可以传递字符串或正则表达式而不是谓词函数。