node.js - 流式传输数据并写入 MongoDB

Question

刚开始调查从一台服务器到另一台服务器的流数据。我正在使用 REQUEST，我已经能够将数据从一个文件 PUT 到另一个服务器上的另一个文件，如下所示:

...
push: function(argv) {
  fs.createReadStream('source_test.json').pipe(request.put('https://path/to/server/'));
}
...

和

//server-side
router.put('/', function (req, res, next) {
  req.pipe(fs.createWriteStream('dest_test.json'));
  res.status(200).end();
});

这很好用。现在我想将内容写入 MongoDB 而不是文件。我很好地将“普通数据”写入 MongoDB，但我想知道将流数据写入数据库的典型方法是什么。我假设我必须等到流式传输完成然后写入数据库？

任何建议，包括使代码更健壮的建议，我们将不胜感激。

Answer 1

假设您以 JSON 文档的形式传输数据。

如果您正在流式传输单个 JSON 文档，那么您必须等待它到达，然后再将其传递并发送到 MongoDB 驱动程序。

但如果您正在流式传输一组文档，那么您应该能够在第一个文档到达后立即开始处理它们。

首先，有两种方法可以对大量的 JSON 文档进行 steam:

• 使用换行符分隔的 JSON 文档列表(即每一行都是一个有效的 JSON 文档)
• 使用单个 JSON 数组

解析换行分隔的JSON数据

你可以用 event-stream module 解析它:

var es = require('event-stream')

req
  .pipe(es.split())
  .pipe(es.parse())
  .pipe(es.map(function (doc, next) {
    collection.insert(doc, next);
  }))

您可以使用相同的模块从 MongoDB 流式传输数据。这是使用 mongoose 执行此操作的示例:

collection
  .find()
  .stream()
  .pipe(es.stringify())
  .pipe(request.put(uri))

解析JSON数组

从流中解析 JSON 数组是一项更困难的任务，但您可以使用 JSONStream module 来完成它:

req
  .pipe(JSONStream.parse('*'))
  .pipe(es.map(function (doc, next) {
    collection.insert(doc, next);
  }))

并从 MongoDB 流式传输数据:

collection
  .find()
  .stream()
  .pipe(JSONStream.stringify())
  .pipe(request.put(uri))

N.B. 确保您没有使用 [body-parser 中间件] 或任何其他会尝试为您解析传入的 JSON 数据的中间件，因为它'将抵消使用 JSONStream 的所有好处。