在 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

截至 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 2

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

原来有。

• JSONStream “流式处理 JSON.parse 和 stringify”

由于我刚找到它，我显然没有使用它，所以我无法评论它的质量，但我很想知道它是否有效。

考虑以下 Javascript 和 _.isString：

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

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

Answer 3

我有类似的要求，我需要在节点 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 4

我写了一个可以做到这一点的模块，叫做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. 一个选项对象，指示输入是换行符分隔的 JSON（这是处理问题中的格式 B，格式 A 不需要）。

在被调用时，bfj.match 将深度优先解析输入流中的 JSON，使用每个值调用谓词以确定是否将该项目推送到结果流。谓词传递三个参数：

1. 属性键或数组索引（顶级项为undefined）。

2. 值本身。

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

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

Answer 5

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

[
   {"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 6

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

我进行了两次测试，以查看从 81MB geojson 文件中打印出每个特征的属性时的性能。

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

在第二个测试中，我使用了@arcseldon 的答案和 JSONStream + event-stream。我修改了 JSONPath 查询以仅选择我需要的内容。这次内存从未超过 82MB，但是，整个过程现在需要 70 秒才能完成！

Answer 7

我使用split npm module 解决了这个问题。通过管道将您的流分成拆分，它将“拆分一个流并重新组合它，以便每一行都是一个块”。

示例代码：

var fs = require('fs')
  , split = require('split')
  ;

var stream = fs.createReadStream(filePath, {flags: 'r', encoding: 'utf-8'});
var lineStream = stream.pipe(split());
linestream.on('data', function(chunk) {
    var json = JSON.parse(chunk);           
    // ...
});

Answer 8

要逐行处理文件，您只需将文件的读取和作用于该输入的代码解耦。您可以通过缓冲输入来完成此操作，直到您点击换行符。假设我们每行有一个 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 vs CRLF），然后在一行中调用JSON.parse。此时，您可以对您的对象做任何您需要的事情。

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

因为一次不会超过一块数据在内存中，这将是非常节省内存的。它也将非常快。快速测试显示我在 15 毫秒内处理了 10,000 行。

Answer 9

我认为您需要使用数据库。在这种情况下，MongoDB 是一个不错的选择，因为它与 JSON 兼容。

更新： 您可以使用mongoimport工具将JSON数据导入MongoDB。

mongoimport --collection collection --file collection.json