Node.js:从MongoDB到文件的流式传输

这是因为find().stream()流式传输对象，而Json2csvStream需要字符串。事件流可以帮助您将对象字符串化。我还简化了你的代码，有一些不必要的东西：

var fs = require('fs');
var MongoClient = require('mongodb').MongoClient;
var es = require('event-stream');
var Json2csvStream = require('json2csv-stream');
// var Db = require('mongodb').Db;
// var Server = require('mongodb').Server;
// var Stream = require('stream');
// var JSONStream = require('JSONStream');
// var csv = require('csv');
var fields = ['execAmendTime', 'execTime', 'commissionUSD', 'region'];
// Connect to the db
// you can put the db name in the url
MongoClient.connect("mongodb://localhost:27017/test_db", function (err, db) {
    if (err) {
        return console.dir(err);
    } else {
        console.log("We are connected");
    }
    // without strict: true, err is always null
    // in strict mode, there is an err if the collection doesn't exist
    db.collection('test', { strict: true }, function (err, collection) {
        if (err) {
            return console.dir(err);
        }
        var json2csv = new Json2csvStream();
        var writer = fs.createWriteStream('out.csv');
        var mongoStream = collection.find(
            { tradeDate: new Date('2015-12-29T00:00:00.000Z') }
        ).stream();
        var stream = mongoStream
            .pipe(es.map(function (doc, next) {
                doc = JSON.stringify(doc);
                // console.log(doc);
                next(null, doc);
            })).pipe(json2csv).pipe(writer).on('close', function () {
                console.log('done...');
                db.close();
            });
    });
});

@Shanoor的回答对我帮助很大。这是我修改的（通用）脚本，用于将MongoDB光标/流输出写入gzcsv文件。

const fs = require("fs");
const csvStringify = require("csv-stringify");
const { Transform } = require('stream');
const zlib = require("zlib");
const { MongoClient } = require('mongodb');
const url = 'mongodb://localhost:27017';
const client = new MongoClient(url);
const dbName = 'myProject'; // your DB name
// input the file name on command prompt
const gzFilePath = process.argv[2];
(async () => {
  // Use connect method to connect to the server
  await client.connect();
  console.log('Connected successfully to server');
  const db = client.db(dbName);
  // exucute the query
  const cursor = await db.collection(/*collection name*/'user')
    .find({/* your query*/});
  // write stream to input gz file name
  const writeStream = fs.createWriteStream(gzFilePath);
  const stringifier = csvStringify.stringify({
    header: true
  });
  writeStream.on('finish', () => {
    console.log('All filtered records written.');
  });
  writeStream.on('error', err => {
    // handle the error properly here
    throw error;
  });
  const gz = zlib.createGzip();
  // process the data with stream
  cursor.stream()
    .pipe(handleStream)
    .pipe(stringifier)
    .pipe(gz)
    .pipe(writeStream);
})();
const handleStream = new Transform({
  readableObjectMode: true,
  writableObjectMode: true,
  transform (chunk, encoding, callback) {
    // process or filter the document here
    const {_id, name, age} = chunk;
    if (age > 18) {
      this.push({_id, name, age});
    }
    callback();
  }
});