在节点 JS 中缓冲流中的数据以执行批量插入答案

【问题标题】：Bufferizing data from stream in nodeJS for perfoming bulk insert在节点 JS 中缓冲流中的数据以执行批量插入
【发布时间】：2021-02-21 00:37:00
【问题描述】：

如何在 nodeJS 中有效地缓冲从流到批量插入的事件，而不是从流中接收到的每条记录的唯一插入。这是我想到的伪代码：

// Open MongoDB connection

mystream.on('data', (record) => {
   // bufferize data into an array
   // if the buffer is full (1000 records)
   // bulk insert into MongoDB and empty buffer
})

mystream.on('end', () => {
   // close connection
})

这看起来真实吗？有没有可能的优化？现有的图书馆能提供这样的便利吗？

【问题讨论】：

nodejs 原生的 stream api 听起来很合适，你应该考虑使用 Writable。缓冲区的大小可以通过设置 highWaterMark 来控制。可写类有一个final() 函数，一旦流完成就会调用该函数。这可以用来关闭数据库连接。
感谢您的回答，我也考虑过该选项，这可能是解决该问题的最佳方法，您输入的数据越多，放入缓冲区的数据越多，您收到的数据就越多在缓冲区中将自动填充 MongoDB 数据库，我假设您还可以通过这种方式控制数据流，并使进入输入的数据自动销毁。我计划通过这种方法使用从小到大的数据集（从几 kb 到 5-10gb 的流数据）
MongoDB 的本机驱动程序（和 Mongoose API）都公开了一个 DB 游标接口，该接口可以封装为 stream.Readable (stream.Readable.from())，然后通过管道传输到缓冲区 Writable。因此，脚本获取的数据不会超过它可以存储在其可写缓冲区中的数据。
这个例子非常接近我正在寻找的github.com/sorribas/mongo-write-stream/blob/master/index.js

标签： javascript node.js mongodb stream buffer

【解决方案1】：

使用 NodeJS 的 stream 库，可以简洁高效地实现为：

const stream = require('stream');
const util = require('util');
const mongo = require('mongo');

const streamSource; // A stream of objects from somewhere

// Establish DB connection
const client = new mongo.MongoClient("uri");
await client.connect();

// The specific collection to store our documents
const collection = client.db("my_db").collection("my_collection");

await util.promisify(stream.pipeline)( 
  streamSource, 
  stream.Writable({
    objectMode: true,
    highWaterMark: 1000,
    writev: async (chunks, next) => {
      try {
        const documents = chunks.map(({chunk}) => chunk);
        
        await collection.insertMany(docs, {ordered: false});

        next();
      }
      catch( error ){
        next( error );
      }
    }
  })
);

【讨论】：

再次，非常感谢您的帮助，这可能是我的问题的解决方案。
我对 Nodejs Streams 比较陌生（在更复杂的用例中）。这个答案对我帮助很大！谢谢大家。

【解决方案2】：

我最终得到了一个不依赖的解决方案。

const { MongoClient } = require("mongodb")
const url = process.env.MONGO_URI || "mongodb://localhost:27019";
const connection = MongoClient.connect(url, { useNewUrlParser: true, useUnifiedTopology: true })
    Promise.resolve(connection)
        .then((db) => {
            const dbName = "databaseName";
            const collection = 'collection';
            const dbo = db.db(dbName);

            let buffer = []

            stream.on("data", (row: any) => {
                buffer.push(row)
                if (buffer.length > 10000) {
                    dbo.collection(collection).insertMany(buffer, {ordered: false});
                    buffer = []
                }
            });

            stream.on("end", () => {
                // insert last chunk
                dbo.collection(collection).insertMany(buffer, {ordered: false})
                    .then(() => {
                        console.log("Done!");
                        db.close();
                    })
                
            });
            sas_stream.on("error", (err) => console.log(err));

        })
        .catch((err) => {
            console.log(err)
        })

【讨论】：